Determine project from credentials. Only require subscription short name

Author: kayleg

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cloud-pubsub"
-version = "0.1.0"
+version = "0.2.0"
 authors = ["Kayle Gishen <k@bkdsw.com>"]
 edition = "2018"
 description = "Google Cloud PubSub Client"

README.md

@@ -57,10 +57,10 @@ fn main() {
 
 ## Subscribing
 
-The full canonical subscription name should be used when subscribing:
+### Connecting to existing subscription
 
 ```
-let sub = my_client.subscribe("projects/{google-project-id}/subscriptions/{subscription-name}")
+let sub = my_client.subscribe("subscription-name")
 ```
 
 Currently a subscription needs to be created using the gcloud CLI or the web interface. There are

src/client.rs

@@ -11,6 +11,7 @@ use tokio::timer::Interval;
 pub struct State {
     token: Option<goauth::auth::Token>,
     credentials_path: String,
+    project: Option<String>,
 }
 
 impl State {
@@ -21,20 +22,24 @@ impl State {
     pub fn access_token(&self) -> &str {
         self.token.as_ref().unwrap().access_token()
     }
+
+    pub fn project(&self) -> &str {
+        &(self.project.as_ref().expect("Google Cloud Project has not been set. If it is not in your credential file, call set_project to set it manually."))
+    }
 }
 
 pub trait Client
 where
-    Self: std::marker::Sized + TokenAuthenticated,
+    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;
 
-pub trait TokenAuthenticated {
     fn spawn_token_renew(&self);
     fn refresh_token(&mut self) -> Result<(), error::Error>;
-    fn get_token(&self) -> Result<goauth::auth::Token, goauth::error::GOErr>;
+    fn get_token(&mut self) -> Result<goauth::auth::Token, goauth::error::GOErr>;
 }
 
 pub type BaseClient = Arc<RwLock<State>>;
@@ -43,6 +48,7 @@ impl Client for BaseClient {
     fn subscribe(&self, name: String) -> Subscription {
         Subscription {
             client: self.clone(),
+            canonical_name: format!("projects/{}/subscriptions/{}", self.project(), name),
             name,
         }
     }
@@ -51,16 +57,23 @@ impl Client for BaseClient {
         let mut client = Arc::new(RwLock::new(State {
             token: None,
             credentials_path,
+            project: None,
         }));
 
         match client.refresh_token() {
             Ok(_) => Ok(client),
             Err(e) => Err(e),
         }
     }
-}
 
-impl TokenAuthenticated for BaseClient {
+    fn set_project(&mut self, project: String) {
+        self.write().unwrap().project = Some(project);
+    }
+
+    fn project(&self) -> String {
+        self.read().unwrap().project().to_string()
+    }
+
     fn spawn_token_renew(&self) {
         let mut client = self.clone();
         let renew_token_task = Interval::new(Instant::now(), Duration::from_secs(15 * 60))
@@ -86,11 +99,13 @@ impl TokenAuthenticated for BaseClient {
         }
     }
 
-    fn get_token(&self) -> Result<goauth::auth::Token, goauth::error::GOErr> {
+    fn get_token(&mut self) -> Result<goauth::auth::Token, goauth::error::GOErr> {
         let credentials =
             goauth::credentials::Credentials::from_file(&self.read().unwrap().credentials_path)
                 .unwrap();
 
+        self.set_project(credentials.project());
+
         let claims = JwtClaims::new(
             credentials.iss(),
             &Scope::PubSub,

src/error.rs

@@ -1,10 +1,22 @@
+use serde_derive::Deserialize;
 use std::fmt;
 
+#[derive(Deserialize)]
+#[serde(untagged)]
 pub enum Error {
+    #[serde(skip_deserializing)]
     PubSubAuth(goauth::error::GOErr),
+    #[serde(skip_deserializing)]
     Http(hyper::Error),
+    #[serde(skip_deserializing)]
     Json(serde_json::Error),
+    #[serde(skip_deserializing)]
     Base64(base64::DecodeError),
+    PubSub {
+        code: i32,
+        message: String,
+        status: String,
+    },
 }
 
 impl fmt::Display for Error {
@@ -14,6 +26,15 @@ impl fmt::Display for Error {
             Error::Http(e) => write!(f, "Hyper({})", e),
             Error::Json(e) => write!(f, "Json({})", e),
             Error::Base64(e) => write!(f, "Base64({})", e),
+            Error::PubSub {
+                code,
+                message,
+                status,
+            } => write!(
+                f,
+                "PubSub(code: {}, status: {}, message: {})",
+                code, status, message
+            ),
         }
     }
 }

src/subscription.rs

@@ -11,6 +11,7 @@ use std::sync::{Arc, RwLock};
 struct Response {
     #[serde(alias = "receivedMessages")]
     received_messages: Option<Vec<Message>>,
+    error: Option<error::Error>,
 }
 
 #[derive(Serialize)]
@@ -22,16 +23,20 @@ struct AckRequest {
 pub struct Subscription {
     pub(crate) client: Arc<RwLock<State>>,
     pub name: String,
+    pub(crate) canonical_name: String,
 }
 
 impl Subscription {
     pub fn acknowledge_messages(&self, ids: Vec<String>) -> impl Future<Item = (), Error = ()> {
         let https = HttpsConnector::new(4).unwrap();
         let client = hyper::Client::builder().build::<_, hyper::Body>(https);
 
-        let uri: hyper::Uri = format!("https://pubsub.googleapis.com/v1/{}:acknowledge", self.name)
-            .parse()
-            .unwrap();
+        let uri: hyper::Uri = format!(
+            "https://pubsub.googleapis.com/v1/{}:acknowledge",
+            self.canonical_name
+        )
+        .parse()
+        .unwrap();
 
         let json = serde_json::to_string(&AckRequest { ack_ids: ids }).unwrap();
 
@@ -66,9 +71,12 @@ impl Subscription {
         let https = HttpsConnector::new(4).unwrap();
         let client = hyper::Client::builder().build::<_, hyper::Body>(https);
 
-        let uri: hyper::Uri = format!("https://pubsub.googleapis.com/v1/{}:pull", self.name)
-            .parse()
-            .unwrap();
+        let uri: hyper::Uri = format!(
+            "https://pubsub.googleapis.com/v1/{}:pull",
+            self.canonical_name
+        )
+        .parse()
+        .unwrap();
 
         let json = r#"{"maxMessages": 100}"#;
 
@@ -96,6 +104,9 @@ impl Subscription {
             .from_err::<error::Error>()
             .and_then(|body| {
                 let response: Response = serde_json::from_slice(&body)?;
+                if let Some(e) = response.error {
+                    return Err(e);
+                }
                 let messages = response.received_messages.unwrap_or_default();
                 let ack_ids: Vec<String> = messages
                     .as_slice()