change get_messages signature as discussed (kayleg#2)

* 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 occurred.

Co-authored-by: Robert Walker <robert@walker.st>

Author: hseeberger

examples/long_lived.rs

@@ -27,16 +27,18 @@ fn schedule_pubsub_pull(subscription: Arc<Subscription>) {
     task::spawn(async move {
         while subscription.client().is_running() {
             match subscription.get_messages::<UpdatePacket>().await {
-                Ok((packets, acks)) => {
-                    for packet in packets {
-                        println!("Received: {:?}", packet);
-                    }
-
-                    if !acks.is_empty() {
-                        let s = Arc::clone(&subscription);
-                        task::spawn(async move {
-                            s.acknowledge_messages(acks).await;
-                        });
+                Ok(messages) => {
+                    for (result, ack_id) in messages {
+                        match result {
+                            Ok(message) => {
+                                println!("Received: {:?}", message);
+                                let subscription = Arc::clone(&subscription);
+                                task::spawn(async move {
+                                    subscription.acknowledge_messages(vec![ack_id]).await;
+                                });
+                            }
+                            Err(e) => log::error!("Failed converting to UpdatePacket: {}", e),
+                        }
                     }
                 }
                 Err(e) => eprintln!("Failed to pull PubSub messages: {}", e),

examples/serde.rs

@@ -36,19 +36,22 @@ async fn main() {
         Ok(p) => p,
     };
 
-    let order_sub = Arc::new(pubsub.subscribe(config.pubsub_subscription));
-    match order_sub.clone().get_messages::<UpdatePacket>().await {
-        Ok((packets, acks)) => {
-            for packet in packets {
-                println!("Received: {:?}", packet);
-            }
-
-            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");
+    let subscription = Arc::new(pubsub.subscribe(config.pubsub_subscription));
+    match subscription.get_messages::<UpdatePacket>().await {
+        Ok(messages) => {
+            for (result, ack_id) in messages {
+                match result {
+                    Ok(message) => {
+                        println!("Received: {:?}", message);
+                        let subscription = Arc::clone(&subscription);
+                        task::spawn(async move {
+                            subscription.acknowledge_messages(vec![ack_id]).await;
+                        });
+                    }
+                    Err(e) => log::error!("Failed converting to UpdatePacket: {}", e),
+                }
             }
         }
-        Err(e) => println!("Error Checking PubSub: {}", e),
+        Err(e) => eprintln!("Failed to pull PubSub messages: {}", e),
     }
 }

examples/singleshot.rs

@@ -60,7 +60,7 @@ impl Subscription {
 
     pub async fn get_messages<T: FromPubSubMessage>(
         &self,
-    ) -> Result<(Vec<T>, Vec<String>), error::Error> {
+    ) -> Result<Vec<(Result<T, error::Error>, String)>, error::Error> {
         let client = self
             .client
             .as_ref()
@@ -76,6 +76,7 @@ impl Subscription {
         *req.uri_mut() = uri.clone();
 
         let response = client.hyper_client().request(req).await?;
+
         if response.status() == StatusCode::NOT_FOUND {
             return Err(error::Error::PubSub {
                 code: 404,
@@ -88,24 +89,13 @@ impl Subscription {
         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()
-            .iter()
-            .map(|packet| packet.ack_id.clone())
-            .collect();
-        let packets = messages
+        let messages = response
+            .received_messages
+            .unwrap_or_default()
             .into_iter()
-            .filter_map(|packet| match T::from(packet.message) {
-                Ok(o) => Some(o),
-                Err(e) => {
-                    error!("Failed converting pubsub {}", e,);
-                    None
-                }
-            })
+            .map(|m| (T::from(m.message), m.ack_id))
             .collect();
-
-        Ok((packets, ack_ids))
+        Ok(messages)
     }
 
     pub async fn destroy(self) -> Result<(), error::Error> {