Added Topic object and ability to subscribe to a topic.

kayleg · kayleg · commit 3be2ad89f7b7 · 2019-04-11T22:46:45.000-04:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "cloud-pubsub"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Kayle Gishen <k@bkdsw.com>"]
 edition = "2018"
 description = "Google Cloud PubSub Client"
@@ -24,3 +24,4 @@ serde_json = "1.0"
 base64 = "0.10.1"
 lazy_static = "1.2.0"
 envy = "0.4"
+rand = "0.6"
diff --git a/README.md b/README.md
@@ -63,5 +63,57 @@ fn main() {
 let sub = my_client.subscribe("subscription-name")
 ```
 
-Currently a subscription needs to be created using the gcloud CLI or the web interface. There are
-plans to support creating a subscription directly from a topic name.
+### Subscribing to a topic
+
+When subscribing to a topic, a random subscription name will be generated. To prevent dangling
+subscriptions, you need to explicitly call `subscription.destroy()`.
+
+```
+let topic = Arc::new(pubsub.topic("update-topic".to_owned()));
+tokio::run(lazy(move || {
+    topic
+        .subscribe()
+        .map(|subscription| {
+            println!("Subscribed to topic with: {}", subscription.name);
+            let sub = Arc::new(subscription);
+            let get = sub
+                .clone()
+                .get_messages::<UpdatePacket>()
+                .map(move |(packets, acks)| {
+                    for packet in packets {
+                        println!("Received: {:?}", packet);
+                    }
+
+                    if !acks.is_empty() {
+                        tokio::spawn(sub.acknowledge_messages(acks));
+                    } else {
+                        println!("Cleaning up");
+                        if let Ok(s) = Arc::try_unwrap(sub) {
+                            let destroy = s
+                                .destroy()
+                                .map(|_r| println!("Successfully deleted subscription"))
+                                .map_err(|e| eprintln!("Failed deleting subscription: {}", e))
+                                .then(move |_| {
+                                    drop(pubsub);
+                                    Ok(())
+                                });
+                            tokio::spawn(destroy);
+                        } else {
+                            eprintln!("Subscription is still owned");
+                        }
+                    }
+                })
+                .map_err(|e| println!("Error Checking PubSub: {}", e));
+            tokio::spawn(get);
+        })
+        .map_err(|e| eprintln!("Failed to subscribe: {}", e))
+}));
+```
+
+## Known Caveats
+
+A Hyper Client is retained by each PubSub client for reuse. This can cause the tokio runtime to not exit until the Hyper Client keep-alive terminates. To shutdown the runtime immediately on complete, the PubSub client needs to be explicitly dropped.
+
+```
+drop(my_client);
+```
diff --git a/examples/serde.rs b/examples/serde.rs
@@ -1,5 +1,5 @@
 use cloud_pubsub::error;
-use cloud_pubsub::{BaseClient, Client, EncodedMessage, FromPubSubMessage};
+use cloud_pubsub::{Client, EncodedMessage, FromPubSubMessage};
 use futures::future::lazy;
 use serde_derive::Deserialize;
 use std::sync::Arc;
@@ -34,7 +34,7 @@ fn main() {
     }
     let config = parsed_env.unwrap();
 
-    let pubsub = match BaseClient::create(config.google_application_credentials) {
+    let pubsub = match Client::new(config.google_application_credentials) {
         Err(e) => panic!("Failed to initialize pubsub: {}", e),
         Ok(p) => p,
     };
@@ -53,6 +53,7 @@ fn main() {
                 if !acks.is_empty() {
                     tokio::spawn(order_sub.acknowledge_messages(acks));
                 }
+                drop(pubsub);
             })
             .map_err(|e| println!("Error Checking PubSub: {}", e))
     }));
diff --git a/examples/singleshot.rs b/examples/singleshot.rs
@@ -1,5 +1,5 @@
 use cloud_pubsub::error;
-use cloud_pubsub::{BaseClient, Client, EncodedMessage, FromPubSubMessage};
+use cloud_pubsub::{Client, EncodedMessage, FromPubSubMessage};
 use futures::future::lazy;
 use serde_derive::Deserialize;
 use std::sync::Arc;
@@ -31,7 +31,7 @@ fn main() {
     }
     let config = parsed_env.unwrap();
 
-    let pubsub = match BaseClient::create(config.google_application_credentials) {
+    let pubsub = match Client::new(config.google_application_credentials) {
         Err(e) => panic!("Failed to initialize pubsub: {}", e),
         Ok(p) => p,
     };
@@ -50,6 +50,8 @@ fn main() {
                 if !acks.is_empty() {
                     tokio::spawn(order_sub.acknowledge_messages(acks));
                 }
+
+                drop(pubsub);
             })
             .map_err(|e| println!("Error Checking PubSub: {}", e))
     }));
diff --git a/examples/subscribe_to_topic.rs b/examples/subscribe_to_topic.rs
@@ -0,0 +1,79 @@
+use cloud_pubsub::error;
+use cloud_pubsub::{Client, EncodedMessage, FromPubSubMessage};
+use futures::future::lazy;
+use serde_derive::Deserialize;
+use std::sync::Arc;
+use tokio::prelude::*;
+
+#[derive(Deserialize)]
+struct Config {
+    topic: String,
+    google_application_credentials: String,
+}
+
+#[derive(Debug)]
+struct UpdatePacket(String);
+
+impl FromPubSubMessage for UpdatePacket {
+    fn from(message: EncodedMessage) -> Result<Self, error::Error> {
+        match message.decode() {
+            Ok(bytes) => Ok(UpdatePacket(String::from_utf8_lossy(&bytes).into_owned())),
+            Err(e) => Err(error::Error::from(e)),
+        }
+    }
+}
+
+fn main() {
+    let parsed_env = envy::from_env::<Config>();
+    if let Err(e) = parsed_env {
+        eprintln!("ENV is not valid: {}", e);
+        std::process::exit(1);
+    }
+    let config = parsed_env.unwrap();
+
+    let pubsub = match Client::new(config.google_application_credentials) {
+        Err(e) => panic!("Failed to initialize pubsub: {}", e),
+        Ok(p) => p,
+    };
+
+    let topic = Arc::new(pubsub.topic(config.topic));
+
+    tokio::run(lazy(move || {
+        topic
+            .subscribe()
+            .map(|subscription| {
+                println!("Subscribed to topic with: {}", subscription.name);
+                let sub = Arc::new(subscription);
+                let get = sub
+                    .clone()
+                    .get_messages::<UpdatePacket>()
+                    .map(move |(packets, acks)| {
+                        for packet in packets {
+                            println!("Received: {:?}", packet);
+                        }
+
+                        if !acks.is_empty() {
+                            tokio::spawn(sub.acknowledge_messages(acks));
+                        } else {
+                            println!("Cleaning up");
+                            if let Ok(s) = Arc::try_unwrap(sub) {
+                                let destroy = s
+                                    .destroy()
+                                    .map(|_r| println!("Successfully deleted subscription"))
+                                    .map_err(|e| eprintln!("Failed deleting subscription: {}", e))
+                                    .then(move |_| {
+                                        drop(pubsub);
+                                        Ok(())
+                                    });
+                                tokio::spawn(destroy);
+                            } else {
+                                eprintln!("Subscription is still owned");
+                            }
+                        }
+                    })
+                    .map_err(|e| println!("Error Checking PubSub: {}", e));
+                tokio::spawn(get);
+            })
+            .map_err(|e| eprintln!("Failed to subscribe: {}", e))
+    }));
+}
diff --git a/src/client.rs b/src/client.rs
@@ -1,17 +1,23 @@
 use crate::error;
 use crate::subscription::Subscription;
+use crate::topic::Topic;
 use futures::prelude::*;
 use goauth::auth::JwtClaims;
 use goauth::scopes::Scope;
+use hyper::client::HttpConnector;
+use hyper_tls::HttpsConnector;
 use smpl_jwt::Jwt;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 use tokio::timer::Interval;
 
+type HyperClient = Arc<hyper::Client<HttpsConnector<HttpConnector>, hyper::Body>>;
+
 pub struct State {
     token: Option<goauth::auth::Token>,
     credentials_path: String,
     project: Option<String>,
+    hyper_client: HyperClient,
 }
 
 impl State {
@@ -28,53 +34,53 @@ impl State {
     }
 }
 
-pub trait Client
-where
-    Self: std::marker::Sized,
-{
-    fn create(credentials_path: String) -> Result<Self, error::Error>;
-    fn subscribe(&self, name: String) -> Subscription;
-    fn set_project(&mut self, project: String);
-    fn project(&self) -> String;
-
-    fn spawn_token_renew(&self);
-    fn refresh_token(&mut self) -> Result<(), error::Error>;
-    fn get_token(&mut self) -> Result<goauth::auth::Token, goauth::error::GOErr>;
-}
-
-pub type BaseClient = Arc<RwLock<State>>;
+pub struct Client(Arc<RwLock<State>>);
 
-impl Client for BaseClient {
-    fn subscribe(&self, name: String) -> Subscription {
-        Subscription {
-            client: self.clone(),
-            canonical_name: format!("projects/{}/subscriptions/{}", self.project(), name),
-            name,
-        }
+impl Clone for Client {
+    fn clone(&self) -> Self {
+        Client(self.0.clone())
     }
+}
 
-    fn create(credentials_path: String) -> Result<Self, error::Error> {
-        let mut client = Arc::new(RwLock::new(State {
+impl Client {
+    pub fn new(credentials_path: String) -> Result<Self, error::Error> {
+        let mut client = Client(Arc::new(RwLock::new(State {
             token: None,
             credentials_path,
             project: None,
-        }));
+            hyper_client: setup_hyper(),
+        })));
 
         match client.refresh_token() {
             Ok(_) => Ok(client),
             Err(e) => Err(e),
         }
     }
 
-    fn set_project(&mut self, project: String) {
-        self.write().unwrap().project = Some(project);
+    pub fn subscribe(&self, name: String) -> Subscription {
+        Subscription {
+            client: Some(self.clone()),
+            name: format!("projects/{}/subscriptions/{}", self.project(), name),
+            topic: None,
+        }
     }
 
-    fn project(&self) -> String {
-        self.read().unwrap().project().to_string()
+    pub fn set_project(&mut self, project: String) {
+        self.0.write().unwrap().project = Some(project);
     }
 
-    fn spawn_token_renew(&self) {
+    pub fn project(&self) -> String {
+        self.0.read().unwrap().project().to_string()
+    }
+
+    pub fn topic(&self, name: String) -> Topic {
+        Topic {
+            client: Some(Client(self.0.clone())),
+            name: format!("projects/{}/topics/{}", self.project(), name),
+        }
+    }
+
+    pub fn spawn_token_renew(&self) {
         let mut client = self.clone();
         let renew_token_task = Interval::new(Instant::now(), Duration::from_secs(15 * 60))
             .for_each(move |_instant| {
@@ -89,10 +95,10 @@ impl Client for BaseClient {
         tokio::spawn(renew_token_task);
     }
 
-    fn refresh_token(&mut self) -> Result<(), error::Error> {
+    pub fn refresh_token(&mut self) -> Result<(), error::Error> {
         match self.get_token() {
             Ok(token) => {
-                self.write().unwrap().token = Some(token);
+                self.0.write().unwrap().token = Some(token);
                 Ok(())
             }
             Err(e) => Err(error::Error::from(e)),
@@ -101,7 +107,7 @@ impl Client for BaseClient {
 
     fn get_token(&mut self) -> Result<goauth::auth::Token, goauth::error::GOErr> {
         let credentials =
-            goauth::credentials::Credentials::from_file(&self.read().unwrap().credentials_path)
+            goauth::credentials::Credentials::from_file(&self.0.read().unwrap().credentials_path)
                 .unwrap();
 
         self.set_project(credentials.project());
@@ -116,4 +122,41 @@ impl Client for BaseClient {
         let jwt = Jwt::new(claims, credentials.rsa_key().unwrap(), None);
         goauth::get_token_with_creds(&jwt, &credentials)
     }
+
+    pub(crate) fn request<T: Into<hyper::Body>>(
+        &self,
+        method: hyper::Method,
+        data: T,
+    ) -> hyper::Request<hyper::Body>
+    where
+        hyper::Body: std::convert::From<T>,
+    {
+        let mut req = hyper::Request::new(hyper::Body::from(data));
+        *req.method_mut() = method;
+        req.headers_mut().insert(
+            hyper::header::CONTENT_TYPE,
+            hyper::header::HeaderValue::from_static("application/json"),
+        );
+        let readable = self.0.read().unwrap();
+        req.headers_mut().insert(
+            hyper::header::AUTHORIZATION,
+            hyper::header::HeaderValue::from_str(&format!(
+                "{} {}",
+                readable.token_type(),
+                readable.access_token()
+            ))
+            .unwrap(),
+        );
+        req
+    }
+
+    pub fn hyper_client(&self) -> HyperClient {
+        self.0.read().unwrap().hyper_client.clone()
+    }
+}
+
+fn setup_hyper() -> HyperClient {
+    // 4 is number of blocking DNS threads
+    let https = HttpsConnector::new(4).unwrap();
+    Arc::new(hyper::Client::builder().build::<_, hyper::Body>(https))
 }
diff --git a/src/lib.rs b/src/lib.rs
@@ -2,7 +2,9 @@ pub mod client;
 pub mod error;
 pub mod message;
 pub mod subscription;
+pub mod topic;
 
-pub use client::{BaseClient, Client};
+pub use client::Client;
 pub use message::{EncodedMessage, FromPubSubMessage};
 pub use subscription::Subscription;
+pub use topic::Topic;
diff --git a/src/subscription.rs b/src/subscription.rs
diff --git a/src/topic.rs b/src/topic.rs
