implementing Consumer_Update command

Author: DanielePalaia

protocol/src/commands/consumer_update.rs

@@ -0,0 +1,84 @@
+use std::io::Write;
+
+#[cfg(test)]
+use fake::Fake;
+
+use crate::{
+    codec::{Decoder, Encoder},
+    error::{DecodeError, EncodeError},
+    protocol::commands::COMMAND_CONSUMER_UPDATE,
+};
+
+use super::Command;
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct ConsumerUpdateCommand {
+    pub(crate) correlation_id: u32,
+    subscription_id: u8,
+    active: u8,
+}
+
+impl ConsumerUpdateCommand {
+    pub fn new(
+        correlation_id: u32,
+        subscription_id: u8,
+        active: u8,
+    ) -> Self {
+        Self {
+            correlation_id,
+            subscription_id,
+            active,
+        }
+    }
+}
+
+impl Encoder for ConsumerUpdateCommand {
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size()
+            + self.subscription_id.encoded_size()
+            + self.active.encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.subscription_id.encode(writer)?;
+        self.active.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Decoder for ConsumerUpdateCommand {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, subscription_id) = u8::decode(input)?;
+        let (input, active) = u8::decode(input)?;
+
+        Ok((
+            input,
+            ConsumerUpdateCommand {
+                correlation_id,
+                subscription_id,
+                active,
+            },
+        ))
+    }
+}
+
+impl Command for ConsumerUpdateCommand {
+    fn key(&self) -> u16 {
+        COMMAND_CONSUMER_UPDATE
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::commands::tests::command_encode_decode_test;
+
+    use super::{ConsumerUpdateCommand};
+
+    #[test]
+    fn subscribe_request_test() {
+        command_encode_decode_test::<ConsumerUpdateCommand>();
+    }
+}

protocol/src/commands/mod.rs

@@ -29,6 +29,7 @@ pub mod superstream_partitions;
 pub mod superstream_route;
 pub mod tune;
 pub mod unsubscribe;
+pub mod consumer_update;
 
 pub trait Command {
     fn key(&self) -> u16;

protocol/src/protocol.rs

@@ -32,6 +32,9 @@ pub mod commands {
     pub const COMMAND_STREAMS_STATS: u16 = 28;
     pub const COMMAND_CREATE_SUPER_STREAM: u16 = 29;
     pub const COMMAND_DELETE_SUPER_STREAM: u16 = 30;
+    pub const COMMAND_CONSUMER_UPDATE_REQUEST: u16 = 32794;
+
+
 }
 
 // server responses

protocol/src/response/mod.rs

@@ -14,7 +14,7 @@ use crate::{
         publish_error::PublishErrorResponse, query_offset::QueryOffsetResponse,
         query_publisher_sequence::QueryPublisherResponse, sasl_handshake::SaslHandshakeResponse,
         superstream_partitions::SuperStreamPartitionsResponse,
-        superstream_route::SuperStreamRouteResponse, tune::TunesCommand,
+        superstream_route::SuperStreamRouteResponse, tune::TunesCommand, consumer_update::ConsumerUpdateCommand
     },
     error::DecodeError,
     protocol::commands::*,
@@ -72,6 +72,7 @@ pub enum ResponseKind {
     ExchangeCommandVersions(ExchangeCommandVersionsResponse),
     SuperStreamPartitions(SuperStreamPartitionsResponse),
     SuperStreamRoute(SuperStreamRouteResponse),
+    ConsumerUpdate(ConsumerUpdateCommand),
 }
 
 impl Response {
@@ -107,6 +108,9 @@ impl Response {
             ResponseKind::SuperStreamRoute(super_stream_route_command) => {
                 Some(super_stream_route_command.correlation_id)
             }
+            ResponseKind::ConsumerUpdate(consumer_update_command) => {
+                Some(consumer_update_command.correlation_id)
+            }
         }
     }
 
@@ -188,6 +192,8 @@ impl Decoder for Response {
                 .map(|(remaining, kind)| (remaining, ResponseKind::SuperStreamPartitions(kind)))?,
             COMMAND_ROUTE => SuperStreamRouteResponse::decode(input)
                 .map(|(remaining, kind)| (remaining, ResponseKind::SuperStreamRoute(kind)))?,
+            COMMAND_CONSUMER_UPDATE => ConsumerUpdateCommand::decode(input)
+                .map(|(remaining, kind)| (remaining, ResponseKind::ConsumerUpdate(kind)))?,
             n => return Err(DecodeError::UnsupportedResponseType(n)),
         };
         Ok((input, Response { header, kind }))
@@ -228,22 +234,24 @@ mod tests {
             query_publisher_sequence::QueryPublisherResponse,
             sasl_handshake::SaslHandshakeResponse,
             superstream_partitions::SuperStreamPartitionsResponse,
-            superstream_route::SuperStreamRouteResponse, tune::TunesCommand,
+            superstream_route::SuperStreamRouteResponse, tune::TunesCommand, consumer_update::ConsumerUpdateCommand
         },
         protocol::{
             commands::{
                 COMMAND_CLOSE, COMMAND_DELIVER, COMMAND_HEARTBEAT, COMMAND_METADATA,
                 COMMAND_METADATA_UPDATE, COMMAND_OPEN, COMMAND_PARTITIONS, COMMAND_PEER_PROPERTIES,
                 COMMAND_PUBLISH_CONFIRM, COMMAND_PUBLISH_ERROR, COMMAND_QUERY_OFFSET,
                 COMMAND_QUERY_PUBLISHER_SEQUENCE, COMMAND_ROUTE, COMMAND_SASL_AUTHENTICATE,
-                COMMAND_SASL_HANDSHAKE, COMMAND_TUNE,
+                COMMAND_SASL_HANDSHAKE, COMMAND_TUNE, COMMAND_CONSUMER_UPDATE,
             },
             version::PROTOCOL_VERSION,
         },
         response::COMMAND_EXCHANGE_COMMAND_VERSIONS,
         types::Header,
         ResponseCode,
     };
+    use crate::protocol::commands::COMMAND_CONSUMER_UPDATE;
+
     impl Encoder for ResponseKind {
         fn encoded_size(&self) -> u32 {
             match self {
@@ -273,6 +281,9 @@ mod tests {
                 ResponseKind::SuperStreamRoute(super_stream_response) => {
                     super_stream_response.encoded_size()
                 }
+                ResponseKind::ConsumerUpdate(consumer_update_response) => {
+                    consumer_update_response.encoded_size()
+                }
             }
         }
 
@@ -307,6 +318,9 @@ mod tests {
                 ResponseKind::SuperStreamRoute(super_stream_command_versions) => {
                     super_stream_command_versions.encode(writer)
                 }
+                ResponseKind::ConsumerUpdate(consumer_update_command_version) => {
+                    consumer_update_command_version.encode(writer)
+                }
             }
         }
     }
@@ -494,4 +508,13 @@ mod tests {
             COMMAND_ROUTE
         );
     }
+
+    #[test]
+    fn consumer_update_response_test() {
+        response_test!(
+            ConsumerUpdateCommand,
+            ResponseKind::ConsumerUpdate,
+            COMMAND_CONSUMER_UPDATE
+        );
+    }
 }