Added example of long lived process

Author: kayleg

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cloud-pubsub"
-version = "0.3.0"
+version = "0.3.1"
 authors = ["Kayle Gishen <k@bkdsw.com>"]
 edition = "2018"
 description = "Google Cloud PubSub Client"
@@ -25,3 +25,4 @@ base64 = "0.10.1"
 lazy_static = "1.2.0"
 envy = "0.4"
 rand = "0.6"
+ctrlc = { git = "https://github.com/kayleg/rust-ctrlc" }

examples/long_lived.rs

@@ -0,0 +1,108 @@
+use cloud_pubsub::error;
+use cloud_pubsub::{Client, EncodedMessage, FromPubSubMessage, Subscription};
+use ctrlc;
+use futures::future::lazy;
+use serde_derive::Deserialize;
+use std::sync::Arc;
+use std::time::Duration;
+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 schedule_pubsub_pull(subscription: Arc<Subscription>) {
+    let sub = Arc::clone(&subscription);
+    let get = subscription
+        .clone()
+        .get_messages::<UpdatePacket>()
+        .map(move |(packets, acks)| {
+            for packet in packets {
+                println!("Received: {:?}", packet);
+            }
+
+            if !acks.is_empty() {
+                tokio::spawn(subscription.acknowledge_messages(acks));
+            }
+        })
+        .map_err(|e| println!("Error Checking PubSub: {}", e))
+        .and_then(|_| {
+            if sub.client().is_running() {
+                schedule_pubsub_pull(sub);
+            } else {
+                println!("No longer pulling");
+            }
+
+            Ok(())
+        });
+    tokio::spawn(get);
+}
+
+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) => Arc::new(p),
+    };
+
+    let topic = Arc::new(pubsub.topic(config.topic));
+
+    tokio::run(lazy(move || {
+        pubsub.spawn_token_renew(Duration::from_secs(2));
+
+        topic
+            .subscribe()
+            .map(move |subscription| {
+                println!("Subscribed to topic with: {}", subscription.name);
+                let sub = Arc::new(subscription);
+
+                schedule_pubsub_pull(Arc::clone(&sub));
+
+                ctrlc::set_once_handler(move || {
+                    println!("Cleaning up");
+                    pubsub.stop();
+                    let client = Arc::clone(&pubsub);
+                    println!("Waiting for current Pull to finish....");
+                    while Arc::strong_count(&sub) > 1 {}
+                    println!("Deleting subscription");
+                    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 |_| {
+                                if let Ok(c) = Arc::try_unwrap(client) {
+                                    drop(c);
+                                }
+                                Ok(())
+                            });
+                        tokio::run(destroy);
+                    } else {
+                        eprintln!("Subscription was still ownded");
+                    }
+                })
+                .expect("Error setting Ctrl-C handler");
+            })
+            .map_err(|e| eprintln!("Failed to subscribe: {}", e))
+    }));
+}

src/client.rs

@@ -1,12 +1,14 @@
 use crate::error;
 use crate::subscription::Subscription;
 use crate::topic::Topic;
+use futures::future;
 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::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 use tokio::timer::Interval;
@@ -18,6 +20,7 @@ pub struct State {
     credentials_path: String,
     project: Option<String>,
     hyper_client: HyperClient,
+    running: Arc<AtomicBool>,
 }
 
 impl State {
@@ -49,6 +52,7 @@ impl Client {
             credentials_path,
             project: None,
             hyper_client: setup_hyper(),
+            running: Arc::new(AtomicBool::new(true)),
         })));
 
         match client.refresh_token() {
@@ -80,9 +84,23 @@ impl Client {
         }
     }
 
-    pub fn spawn_token_renew(&self) {
+    pub fn is_running(&self) -> bool {
+        self.0.read().unwrap().running.load(Ordering::SeqCst)
+    }
+
+    pub fn stop(&self) {
+        self.0
+            .write()
+            .unwrap()
+            .running
+            .store(false, Ordering::SeqCst)
+    }
+
+    pub fn spawn_token_renew(&self, interval: Duration) {
         let mut client = self.clone();
-        let renew_token_task = Interval::new(Instant::now(), Duration::from_secs(15 * 60))
+        let c = self.clone();
+        let renew_token_task = Interval::new(Instant::now(), interval)
+            .take_while(move |_| future::ok(c.is_running()))
             .for_each(move |_instant| {
                 println!("Renewing pubsub token");
                 if let Err(e) = client.refresh_token() {

src/subscription.rs

@@ -118,4 +118,8 @@ impl Subscription {
             .and_then(|_res| Ok(()))
             .from_err::<error::Error>()
     }
+
+    pub fn client(&self) -> &Client {
+        self.client.as_ref().unwrap()
+    }
 }