Separate services from connectors

Previously the library attempted to abstract over different HTTP(s)
connectors by passing them as generics through various types. Although
this allows certain flexibility (e.g. choosing openssl vs rustls), it
was probably the wrong abstraction layer in the first place.

This change moves most of the genericism to the grpc-service layer for
grpc services (bigtable and pubsub). The library's "common" path via
builders will provide a sensible default implementation, but it will now
be possible to substitute the entire tower::service stack provided to
the grpc API. This is more flexible than the prior connector
abstraction, but also harder to use (you basically have to build
everything yourself). This feels like the right balance between a
batteries-included default and a build-your-own option.

Along the way I've also made some updates to non-exhaustivity in
generated code, to help ease future changes without breaking semver.
This is unfortunately less ergonomic, but not _too_ bad. Future changes
may include some builder pattern to make protobuf construction easier.

The GCS API is still largely the same, however it is still not in an
ideal place. Future changes may bring it more towards service
abstraction.

Supersedes standard-ai#30
Fixes standard-ai#35

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ya-gcp"
-version = "0.10.0"
+version = "0.11.0"
 authors = ["Renar Narubin <[email protected]>"]
 edition = "2021"
 description = "APIs for using Google Cloud Platform services"
@@ -30,10 +30,10 @@ path = "examples/bigtable.rs"
 required-features = ["bigtable"]
 
 [features]
-default = ["rustls"]
+default = ["rustls-native-certs"]
 
-rustls = ["hyper-rustls"]
-openssl = ["hyper-openssl"] # TODO maybe should be native-tls instead?
+rustls-native-certs = ["dep:rustls-native-certs", "tonic?/tls-roots"]
+webpki-roots = ["dep:webpki-roots", "tonic?/tls-webpki-roots"]
 
 # an internal feature used by services running grpc
 grpc = ["tonic", "prost", "prost-types", "tower", "derive_more"]
@@ -51,27 +51,29 @@ futures = "0.3"
 http = "0.2"
 humantime-serde = "1"
 hyper = "0.14"
+hyper-rustls = "0.24.2"
 paste = "1"
 rand = "0.8"
+rustls = "0.21.8"
 serde = { version = "1", features = ["derive"] }
 thiserror = "1"
 tokio = { version = "1", features = ["time"] }
-tracing = "0.1"
-yup-oauth2 = "8.1"
+tracing = "0.1.37"
+yup-oauth2 = "8.3.0"
 
 async-stream = { version = "0.3", optional = true }
 async-channel = { version = "1", optional = true }
 derive_more = { version = "0.99", optional = true }
-hyper-openssl = { version = "0.9", optional = true }
-hyper-rustls = { version = "0.22", features = ["rustls-native-certs"], optional = true }
 pin-project = { version = "1.0.11", optional = true }
-prost = { version = "0.11", optional = true }
-prost-types = { version = "0.11", optional = true }
+prost = { version = "0.12.3", optional = true }
+prost-types = { version = "0.12.3", optional = true }
+rustls-native-certs = { version = "0.6.3", optional = true }
 tame-gcs = { version = "0.10.0", optional = true }
 tempdir = { version = "0.3", optional = true }
-tonic = { version = "0.9", optional = true }
+tonic = { version = "0.10.2", optional = true }
 tower = { version = "0.4", features = ["make"], optional = true }
-uuid = { version = "0.8.1", features = ["v4"], optional = true }
+uuid = { version = "1.6", features = ["v4"], optional = true }
+webpki-roots = { version = "0.25.3", optional = true }
 
 [dev-dependencies]
 approx = "0.5"
@@ -84,12 +86,6 @@ tokio = { version = "1.4.0", features = ["rt-multi-thread", "time", "test-util"]
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tracing-tree = "0.2"
 
-[package.metadata.cargo-udeps.ignore]
-# hyper-openssl is only used (and thus detected) if the feature "openssl" is
-# enabled _and_ "rustls" is disabled. CI builds with --all-features, so udeps
-# fails unless we ignore the package explicitly
-normal = ["hyper-openssl"]
-
 [package.metadata.docs.rs]
 rustdoc-args = ["--cfg", "docsrs"]
 all-features = true

examples/bigtable.rs

@@ -32,6 +32,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         AuthFlow::NoAuth
     };
 
+    println!("Creating clients");
+
     let config = ClientBuilderConfig::new().auth_flow(auth);
     let builder = ClientBuilder::new(config).await?;
 
@@ -51,6 +53,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .await?;
 
+    println!("Creating table `{}`", &args.table_name);
+
     match admin
         .create_table(
             &args.table_name,
@@ -69,13 +73,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     }
 
+    println!("Reading tables");
+
     let tables: Vec<_> = admin.list_tables().await?.collect().await;
+
     println!("got tables {:?}", tables);
 
     let mut client = builder
         .build_bigtable_client(bigtable_config, &args.project_name, &args.instance_name)
         .await?;
 
+    println!("setting data");
+
     client
         .set_row_data(
             &args.table_name,
@@ -84,6 +93,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             [("col1", "value"), ("col2", "value")],
         )
         .await?;
+
     println!("set data done");
     println!(
         "all data: {:?}",

examples/pubsub_stream.rs

@@ -58,31 +58,38 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("Creating topic {}", &topic_name);
 
     publisher
-        .create_topic(pubsub::api::Topic {
-            name: topic_name.clone().into(),
-            ..pubsub::api::Topic::default()
+        .raw_api_mut()
+        .create_topic({
+            let mut t = pubsub::api::Topic::default();
+            t.name = topic_name.clone().into();
+            t
         })
         .await?;
 
     println!("Creating subscription {}", &subscription_name);
 
     subscriber
-        .create_subscription(pubsub::api::Subscription {
-            name: subscription_name.clone().into(),
-            topic: topic_name.clone().into(),
-            ..pubsub::api::Subscription::default()
+        .raw_api_mut()
+        .create_subscription({
+            let mut s = pubsub::api::Subscription::default();
+            s.name = subscription_name.clone().into();
+            s.topic = topic_name.clone().into();
+            s
         })
         .await?;
 
     println!("Publishing messages to topic");
 
-    futures::stream::iter(0u32..100)
-        .map(|i| pubsub::api::PubsubMessage {
-            data: format!("message-{}", i).into(),
-            ..pubsub::api::PubsubMessage::default()
+    futures::stream::iter(0u32..15)
+        .map(|i| {
+            let mut m = pubsub::api::PubsubMessage::default();
+            let payload = format!("message-{:02}", i);
+            println!("Sending `{payload}`");
+            m.data = payload.into();
+            m
         })
         .map(Ok)
-        .forward(publisher.publish_topic_sink(topic_name.clone()))
+        .forward(publisher.publish_topic_sink(topic_name.clone(), pubsub::PublishConfig::default()))
         .await?;
 
     println!("Reading back published messages");
@@ -93,32 +100,49 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     );
     futures::pin_mut!(read_stream);
 
-    for i in 0u32..100 {
+    let mut messages = Vec::new();
+    for _ in 0u32..15 {
         let (ack_token, message) = read_stream
             .next()
             .await
             .ok_or("unexpected end of stream")??;
 
-        ack_token.ack().await?;
+        let payload = std::str::from_utf8(&message.data[..]).unwrap();
+        println!("Received `{payload}`");
+        messages.push(payload.to_owned());
 
-        assert_eq!(message.data, format!("message-{}", i));
+        ack_token.ack().await?;
     }
 
+    messages.sort();
+    assert_eq!(
+        messages,
+        (0..15)
+            .map(|i| format!("message-{:02}", i))
+            .collect::<Vec<_>>()
+    );
+
     println!("All messages matched!");
 
     println!("Deleting subscription {}", &subscription_name);
 
     subscriber
-        .delete_subscription(pubsub::api::DeleteSubscriptionRequest {
-            subscription: subscription_name.into(),
+        .raw_api_mut()
+        .delete_subscription({
+            let mut d = pubsub::api::DeleteSubscriptionRequest::default();
+            d.subscription = subscription_name.into();
+            d
         })
         .await?;
 
     println!("Deleting topic {}", &topic_name);
 
     publisher
-        .delete_topic(pubsub::api::DeleteTopicRequest {
-            topic: topic_name.into(),
+        .raw_api_mut()
+        .delete_topic({
+            let mut d = pubsub::api::DeleteTopicRequest::default();
+            d.topic = topic_name.into();
+            d
         })
         .await?;
 

generators/Cargo.toml

@@ -16,12 +16,12 @@ harness = false
 [dependencies]
 anyhow = "1"
 flate2 = "1"
-prost-build = { version = "0.11", features = ["format"] }
+prost-build = { version = "0.12.3", features = ["format"] }
 reqwest = { version = "0.11", features = ["blocking"] }
 structopt = "0.3"
 tar = "0.4"
 tempfile = "3"
-tonic-build = "0.9"
+tonic-build = "0.10"
 
 [dev-dependencies]
 criterion = { version = "0.3", features = ["html_reports"] }

generators/src/grpc.rs

@@ -44,21 +44,36 @@ fn main() -> Result<(), Error> {
     // the wire
     prost_config.bytes(&["."]);
 
-    // The bigtable docs have doc comments that trigger test failures.
-    // (TODO: in newer versions of prost-build, the `format` option might be enough for this)
-    prost_config.disable_comments(&[
+    // Some docs have doc comments that trigger test failures because they're not actually rust.
+    // Maybe future codegen will be better?
+    // Try to keep some docs by pretending the fields don't exist during doctests
+    for ignored_field in [
         "bigtable.v2.RowFilter.Interleave.filters",
+        "bigtable.v2.ReadRowsRequest.reversed",
+    ] {
+        prost_config.field_attribute(ignored_field, "#[cfg(not(doctest))]");
+    }
+
+    // Other types can't be pretended away because typechecks will fail. just disable the comments
+    prost_config.disable_comments(&[
         "bigtable.v2.RowFilter.sink",
         "iam.v1.Policy",
         "iam.v1.AuditConfig",
         "iam.v1.AuditLogConfig",
-        "type.Expr",
     ]);
 
     // the attributes map tend to have a small number of string keys, which are faster to access
     // using a btree than a hashmap. See the crate's benchmarks
     prost_config.btree_map(&["PubsubMessage.attributes"]);
 
+    // Declare all the generated structs and enums as non_exhaustive.
+    //
+    // This helps to reconcile two distinct semver conventions:
+    // 1. in protobuf, adding fields is a semver minor change
+    // 2. in prost codegen, structs are composed of all-pub fields, therefore adding a new field is
+    //    a semver *major* change
+    prost_config.type_attribute(".", "#[non_exhaustive]");
+
     tonic_build::configure()
         .build_client(true)
         .build_server(false)

src/auth/grpc.rs

@@ -1,5 +1,6 @@
 //! Authorization support for gRPC requests
 
+use crate::auth::Auth;
 use futures::future::BoxFuture;
 use std::{
     sync::Arc,
@@ -45,15 +46,15 @@ use tonic::client::GrpcService;
 /// # };
 /// ```
 #[derive(Clone)]
-pub struct AuthGrpcService<Service, C> {
+pub struct AuthGrpcService<Service> {
     inner: Service,
-    auth: Option<crate::Auth<C>>,
-    scopes: Arc<Vec<String>>,
+    auth: Option<Auth>,
+    scopes: Arc<[String]>,
 }
 
-impl<Service, C> AuthGrpcService<Service, C> {
+impl<Service> AuthGrpcService<Service> {
     /// Wrap the given service to add authorization headers to each request
-    pub fn new<ReqBody>(service: Service, auth: Option<crate::Auth<C>>, scopes: Vec<String>) -> Self
+    pub fn new<ReqBody>(service: Service, auth: Option<Auth>, scopes: Vec<String>) -> Self
     where
         // Generic bounds included on the constructor because having them only on the trait impl
         // doesn't produce good compiler diagnostics
@@ -63,7 +64,7 @@ impl<Service, C> AuthGrpcService<Service, C> {
         Self {
             inner: service,
             auth,
-            scopes: Arc::new(scopes),
+            scopes: Arc::from(scopes),
         }
     }
 }
@@ -93,20 +94,11 @@ where
     Grpc(ServiceErr),
 }
 
-impl<Service, C, ReqBody> GrpcService<ReqBody> for AuthGrpcService<Service, C>
+impl<Service, ReqBody> GrpcService<ReqBody> for AuthGrpcService<Service>
 where
     Service: GrpcService<ReqBody> + Clone + Send + 'static,
     Service::Error: std::error::Error + Send + Sync + 'static,
     Service::Future: Send,
-    C: tower::Service<http::Uri> + Clone + Send + Sync + 'static,
-    C::Response: hyper::client::connect::Connection
-        + tokio::io::AsyncRead
-        + tokio::io::AsyncWrite
-        + Send
-        + Unpin
-        + 'static,
-    C::Future: Send + Unpin + 'static,
-    C::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
     ReqBody: Send + 'static,
 {
     type Error = AuthGrpcError<Service::Error, yup_oauth2::Error>;
@@ -290,8 +282,7 @@ mod test {
             }
         }
 
-        let mut auth_service =
-            AuthGrpcService::<_, crate::DefaultConnector>::new(OkService, None, vec![]);
+        let mut auth_service = AuthGrpcService::<_>::new(OkService, None, vec![]);
 
         let result = auth_service.call(http::request::Request::new(())).await;
 

src/auth/mod.rs

@@ -3,6 +3,10 @@
 #[cfg(feature = "grpc")]
 pub mod grpc;
 
+pub(crate) type Auth = yup_oauth2::authenticator::Authenticator<
+    hyper_rustls::HttpsConnector<hyper::client::HttpConnector>,
+>;
+
 /// Add the given authorization token to the given HTTP request
 ///
 /// Returns an error if the token cannot form a valid HTTP header value.