Add RedisMessenger::add_stream_with_buffer_size

plerkle_messenger/src/redis_messenger.rs

@@ -247,6 +247,34 @@ impl Messenger for RedisMessenger {
         }
     }
 
+    async fn add_stream_with_buffer_size(
+        &mut self,
+        stream_key: &'static str,
+        max_buffer_size: usize,
+    ) -> Result<(), MessengerError> {
+        // Add to streams hashmap.
+        let _result = self.streams.insert(
+            stream_key,
+            RedisMessengerStream {
+                max_len: Some(StreamMaxlen::Approx(max_buffer_size)),
+                local_buffer: LinkedList::new(),
+                local_buffer_total: 0,
+                local_buffer_last_flush: Instant::now(),
+            },
+        );
+
+        // Add stream to Redis.
+        let result: RedisResult<()> = self
+            .connection
+            .xgroup_create_mkstream(stream_key, self.consumer_group_name.as_str(), "$")
+            .await;
+
+        if let Err(e) = result {
+            info!("Group already exists: {:?}", e)
+        }
+        Ok(())
+    }
+
     async fn add_stream(&mut self, stream_key: &'static str) -> Result<(), MessengerError> {
         // Add to streams hashmap.
         let _result = self.streams.insert(