Use XACK in streams example

Author: Terkwood

examples/streams.rs

@@ -82,9 +82,30 @@ fn demo_group_reads(client: &redis::Client) {
             }
 
             for _ in 0..repeat {
-                read_group_records(&ca, *slowness).expect("group read");
-
-                thread::sleep(Duration::from_millis(random_wait_millis(*slowness)))
+                let read_reply = read_group_records(&ca, *slowness).expect("group read");
+
+                // fake some expensive work
+                for StreamKey { key, ids } in read_reply.keys {
+                    for StreamId { id, map: _ } in &ids {
+                        thread::sleep(Duration::from_millis(random_wait_millis(*slowness)));
+                        println!(
+                            "Stream {} ID {} Consumer slowness {} SysTime {}",
+                            key,
+                            id,
+                            slowness,
+                            SystemTime::now()
+                                .duration_since(UNIX_EPOCH)
+                                .expect("time")
+                                .as_millis()
+                        );
+                    }
+
+                    // acknowledge each stream and message ID once all messages are
+                    // correctly processed
+                    let id_strs: Vec<&String> =
+                        ids.iter().map(|StreamId { id, map: _ }| id).collect();
+                    con.xack(key, GROUP_NAME, &id_strs).expect("ack")
+                }
             }
         }))
     }
@@ -219,7 +240,7 @@ fn consumer_name(slowness: u8) -> String {
 const GROUP_NAME: &str = "example-group-aaa";
 
 #[cfg(feature = "streams")]
-fn read_group_records(client: &redis::Client, slowness: u8) -> RedisResult<()> {
+fn read_group_records(client: &redis::Client, slowness: u8) -> RedisResult<StreamReadReply> {
     let mut con = client.get_connection().expect("conn");
 
     let opts = StreamReadOptions::default()
@@ -235,21 +256,7 @@ fn read_group_records(client: &redis::Client, slowness: u8) -> RedisResult<()> {
         )
         .expect("records");
 
-    for StreamKey { key, ids } in srr.keys {
-        for StreamId { id, map: _ } in ids {
-            println!(
-                "Stream {} ID {} Consumer slowness {} SysTime {}",
-                key,
-                id,
-                slowness,
-                SystemTime::now()
-                    .duration_since(UNIX_EPOCH)
-                    .expect("time")
-                    .as_millis()
-            );
-        }
-    }
-    Ok(())
+    Ok(srr)
 }
 
 #[cfg(feature = "streams")]