This fix changes the returned values from get_messages from a tuple of

vectors with ack_ids and message, to a HashMap where ack_id is stored
with a Result that contains either the successfully deserialized message
or the error that occured.

Author: olofwalker

examples/serde.rs

@@ -0,0 +1,58 @@
+use cloud_pubsub::error;
+use cloud_pubsub::{Client, EncodedMessage, FromPubSubMessage};
+use serde_derive::Deserialize;
+use std::sync::Arc;
+use tokio::task;
+
+#[derive(Deserialize)]
+struct Config {
+    pubsub_subscription: String,
+    google_application_credentials: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct UpdatePacket {
+    id: u64,
+    name: String,
+}
+
+impl FromPubSubMessage for UpdatePacket {
+    fn from(message: EncodedMessage) -> Result<Self, error::Error> {
+        match message.decode() {
+            Ok(bytes) => serde_json::from_slice::<UpdatePacket>(&bytes).map_err(error::Error::from),
+            Err(e) => Err(error::Error::from(e)),
+        }
+    }
+}
+
+#[tokio::main]
+async 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).await {
+        Err(e) => panic!("Failed to initialize pubsub: {}", e),
+        Ok(p) => p,
+    };
+
+    let order_sub = Arc::new(pubsub.subscribe(config.pubsub_subscription));
+    match order_sub.clone().get_messages::<UpdatePacket>().await {
+        Ok(packets) => {
+            for packet in &packets {
+                println!("Received: {:?}", packet);
+            }
+
+            let acks: Vec<String> = packets.into_iter().map(|packet| packet.0).collect();
+            if !acks.is_empty() {
+                task::spawn(async move { order_sub.acknowledge_messages(acks).await })
+                    .await // This will block until acknowledgement is complete
+                    .expect("Failed to acknoweldge messages");
+            }
+        }
+        Err(e) => println!("Error Checking PubSub: {}", e),
+    }
+}

examples/subscribe_to_topic.rs

@@ -0,0 +1,55 @@
+use cloud_pubsub::error;
+use cloud_pubsub::{Client, EncodedMessage, FromPubSubMessage};
+use serde_derive::Deserialize;
+
+#[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)),
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    let config: Config = envy::from_env().expect("ENV is not valid");
+
+    let pubsub = Client::new(config.google_application_credentials)
+        .await
+        .expect("Failed to initialize pubsub");
+
+    let topic = pubsub.topic(config.topic);
+
+    let sub = topic.subscribe().await.expect("Failed to subscribe");
+
+    println!("Subscribed to topic with: {}", sub.name);
+    let packets = sub
+        .clone()
+        .get_messages::<UpdatePacket>()
+        .await
+        .expect("Error Checking PubSub");
+
+    for packet in &packets {
+        println!("Received: {:?}", packet);
+    }
+
+    if !packets.is_empty() {
+        let acks = packets.into_iter().map(|packet| packet.0).collect();
+        sub.acknowledge_messages(acks).await;
+    } else {
+        println!("Cleaning up");
+        drop(pubsub);
+        sub.destroy().await.expect("Failed deleting subscription");
+        println!("Successfully deleted subscription");
+    }
+}

src/subscription.rs

@@ -6,6 +6,7 @@ use hyper::{Method, StatusCode};
 use lazy_static::lazy_static;
 use log::error;
 use serde_derive::{Deserialize, Serialize};
+use std::collections::HashMap;
 use std::env;
 
 lazy_static! {