Separate services from connectors

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.