SAC: starting implementation

Author: DanielePalaia

protocol/src/commands/consumer_update.rs

@@ -74,7 +74,7 @@ mod tests {
     use super::ConsumerUpdateCommand;
 
     #[test]
-    fn subscribe_request_test() {
+    fn consumer_update_response_test() {
         command_encode_decode_test::<ConsumerUpdateCommand>();
     }
 }

protocol/src/commands/consumer_update_request.rs

@@ -80,7 +80,7 @@ mod tests {
     use super::ConsumerUpdateRequestCommand;
 
     #[test]
-    fn subscribe_request_test() {
+    fn consumer_update_request_test() {
         command_encode_decode_test::<ConsumerUpdateRequestCommand>();
     }
 }

protocol/src/response/mod.rs

@@ -529,13 +529,4 @@ mod tests {
             COMMAND_CONSUMER_UPDATE
         );
     }
-
-    #[test]
-    fn consumer_update_request_response_test() {
-        response_test!(
-            ConsumerUpdateRequestCommand,
-            ResponseKind::ConsumerUpdateRequest,
-            COMMAND_CONSUMER_UPDATE_REQUEST
-        );
-    }
 }

src/client/dispatcher.rs

@@ -1,5 +1,5 @@
 use futures::Stream;
-use rabbitmq_stream_protocol::Response;
+use rabbitmq_stream_protocol::{Response, ResponseKind};
 use std::sync::{
     atomic::{AtomicBool, AtomicU32, Ordering},
     Arc,
@@ -167,7 +167,10 @@ where
         while let Some(result) = stream.next().await {
             match result {
                 Ok(item) => match item.correlation_id() {
-                    Some(correlation_id) => state.dispatch(correlation_id, item).await,
+                    Some(correlation_id) => match item.kind_ref() {
+                        ResponseKind::ConsumerUpdate(consumer_update) => state.notify(item).await,
+                        _ => state.dispatch(correlation_id, item).await,
+                    },
                     None => state.notify(item).await,
                 },
                 Err(e) => {

src/consumer.rs

@@ -26,11 +26,14 @@ use crate::{
     Client, ClientOptions, Environment, MetricsCollector,
 };
 use futures::{task::AtomicWaker, Stream};
+use rabbitmq_stream_protocol::commands::consumer_update::ConsumerUpdateCommand;
 use rand::rngs::StdRng;
 use rand::{seq::SliceRandom, SeedableRng};
 
 type FilterPredicate = Option<Arc<dyn Fn(&Message) -> bool + Send + Sync>>;
 
+type ConsumerUpdateListener = Option<Arc<dyn Fn(bool, &MessageContext) -> u64 + Send + Sync>>;
+
 /// API for consuming RabbitMQ stream messages
 pub struct Consumer {
     // Mandatory in case of manual offset tracking
@@ -82,6 +85,26 @@ impl FilterConfiguration {
     }
 }
 
+pub struct MessageContext {
+    consumer: Consumer,
+    subscriber_name: String,
+    reference: String,
+}
+
+impl MessageContext {
+    pub fn get_consumer(self) -> Consumer {
+        self.consumer
+    }
+
+    pub fn get_subscriber_name(self) -> String {
+        self.subscriber_name
+    }
+
+    pub fn get_reference(self) -> String {
+        self.reference
+    }
+}
+
 /// Builder for [`Consumer`]
 pub struct ConsumerBuilder {
     pub(crate) consumer_name: Option<String>,
@@ -317,27 +340,27 @@ impl MessageHandler for ConsumerMessageHandler {
     async fn handle_message(&self, item: MessageResult) -> crate::RabbitMQStreamResult<()> {
         match item {
             Some(Ok(response)) => {
-                if let ResponseKind::Deliver(delivery) = response.kind() {
+                if let ResponseKind::Deliver(delivery) = response.kind_ref() {
                     let mut offset = delivery.chunk_first_offset;
 
                     let len = delivery.messages.len();
+                    let d = delivery.clone();
                     trace!("Got delivery with messages {}", len);
 
                     // // client filter
                     let messages = match &self.0.filter_configuration {
                         Some(filter_input) => {
                             if let Some(f) = &filter_input.predicate {
-                                delivery
-                                    .messages
+                                d.messages
                                     .into_iter()
                                     .filter(|message| f(message))
                                     .collect::<Vec<Message>>()
                             } else {
-                                delivery.messages
+                                d.messages
                             }
                         }
 
-                        None => delivery.messages,
+                        None => d.messages,
                     };
 
                     for message in messages {
@@ -363,6 +386,8 @@ impl MessageHandler for ConsumerMessageHandler {
                     // TODO handle credit fail
                     let _ = self.0.client.credit(self.0.subscription_id, 1).await;
                     self.0.metrics_collector.consume(len as u64).await;
+                } else {
+                    println!("other message arrived");
                 }
             }
             Some(Err(err)) => {
@@ -377,6 +402,7 @@ impl MessageHandler for ConsumerMessageHandler {
         Ok(())
     }
 }
+
 /// Envelope from incoming message
 #[derive(Debug)]
 pub struct Delivery {

src/superstream_consumer.rs

@@ -105,6 +105,11 @@ impl SuperStreamConsumerBuilder {
         self.client_provided_name = String::from(name);
         self
     }
+
+    pub fn properties(mut self, properties: HashMap<String, String>) -> Self {
+        self.properties = properties;
+        self
+    }
 }
 
 impl Stream for SuperStreamConsumer {