implementing route and partition commands

Author: DanielePalaia

Dockerfile

@@ -1,6 +1,6 @@
-FROM rabbitmq:3.13-rc-management
+FROM rabbitmq:4.0.1-management
 
 COPY .ci/conf/rabbitmq.conf /etc/rabbitmq/rabbitmq.conf
 COPY .ci/conf/enabled_plugins /etc/rabbitmq/enabled_plugins
 
-COPY .ci/certs /etc/rabbitmq/certs
+COPY .ci/certs /etc/rabbitmq/certs

protocol/src/commands/create_super_stream.rs

@@ -88,7 +88,7 @@ impl Command for CreateSuperStreamCommand {
 
 #[cfg(test)]
 mod tests {
-    use crate::commands::create_stream::CreateStreamCommand;
+
     use crate::commands::tests::command_encode_decode_test;
 
     use super::CreateSuperStreamCommand;

protocol/src/commands/mod.rs

@@ -25,6 +25,8 @@ pub mod sasl_authenticate;
 pub mod sasl_handshake;
 pub mod store_offset;
 pub mod subscribe;
+pub mod superstream_partitions;
+pub mod superstream_route;
 pub mod tune;
 pub mod unsubscribe;
 

protocol/src/commands/superstream_partitions.rs

@@ -0,0 +1,145 @@
+use std::io::Write;
+
+#[cfg(test)]
+use fake::Fake;
+
+use crate::{
+    codec::{Decoder, Encoder},
+    error::{DecodeError, EncodeError},
+    protocol::commands::COMMAND_PARTITIONS,
+    FromResponse, ResponseCode,
+};
+use crate::commands::exchange_command_versions::ExchangeCommandVersion;
+use super::Command;
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct SuperStreamPartitionsRequest {
+    correlation_id: u32,
+    super_stream: String,
+}
+
+impl SuperStreamPartitionsRequest {
+    pub fn new(correlation_id: u32, super_stream: String) -> Self {
+        Self {
+            correlation_id,
+            super_stream,
+        }
+    }
+}
+
+impl Encoder for SuperStreamPartitionsRequest {
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size() + self.super_stream.as_str().encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.super_stream.as_str().encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Decoder for SuperStreamPartitionsRequest {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, super_stream) = Option::decode(input)?;
+
+        Ok((
+            input,
+            SuperStreamPartitionsRequest {
+                correlation_id,
+                super_stream: super_stream.unwrap(),
+            },
+        ))
+    }
+}
+
+impl Command for SuperStreamPartitionsRequest {
+    fn key(&self) -> u16 {
+        COMMAND_PARTITIONS
+    }
+}
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct SuperStreamPartitionsResponse {
+    pub(crate) correlation_id: u32,
+    response_code: ResponseCode,
+    pub streams: Vec<String>,
+}
+
+impl SuperStreamPartitionsResponse {
+    pub fn new(correlation_id: u32, streams: Vec<String>, response_code: ResponseCode) -> Self {
+        Self {
+            correlation_id,
+            response_code,
+            streams,
+        }
+    }
+    pub fn is_ok(&self) -> bool {
+        self.response_code == ResponseCode::Ok
+    }
+}
+
+impl Encoder for SuperStreamPartitionsResponse {
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.response_code.encode(writer)?;
+        self.streams.encode(writer)?;
+        Ok(())
+    }
+
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size()
+            + self.response_code.encoded_size()
+            + self.streams.encoded_size()
+    }
+}
+
+impl Decoder for SuperStreamPartitionsResponse {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, response_code) = ResponseCode::decode(input)?;
+        let (input, streams) = <Vec<String>>::decode(input)?;
+
+        Ok((
+            input,
+            SuperStreamPartitionsResponse {
+                correlation_id,
+                response_code,
+                streams,
+            },
+        ))
+    }
+}
+
+impl FromResponse for SuperStreamPartitionsResponse {
+    fn from_response(response: crate::Response) -> Option<Self> {
+        match response.kind {
+            crate::ResponseKind::SuperStreamPartitions(partitions_response) => {
+                Some(partitions_response)
+            }
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use crate::commands::tests::command_encode_decode_test;
+
+    use super::SuperStreamPartitionsRequest;
+    use super::SuperStreamPartitionsResponse;
+
+    #[test]
+    fn super_stream_partition_request_test() {
+        command_encode_decode_test::<SuperStreamPartitionsRequest>();
+    }
+
+    #[test]
+    fn super_stream_partition_response_test() {
+        command_encode_decode_test::<SuperStreamPartitionsResponse>();
+    }
+}

protocol/src/commands/superstream_route.rs

@@ -0,0 +1,150 @@
+use std::io::Write;
+
+#[cfg(test)]
+use fake::Fake;
+
+use crate::{
+    codec::{Decoder, Encoder},
+    error::{DecodeError, EncodeError},
+    protocol::commands::COMMAND_ROUTE,
+    FromResponse, ResponseCode,
+};
+
+use super::Command;
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct SuperStreamRouteRequest {
+    correlation_id: u32,
+    routing_key: String,
+    super_stream: String,
+}
+
+impl SuperStreamRouteRequest {
+    pub fn new(correlation_id: u32, routing_key: String, super_stream: String) -> Self {
+        Self {
+            correlation_id,
+            routing_key,
+            super_stream,
+        }
+    }
+}
+
+impl Encoder for SuperStreamRouteRequest {
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size()
+            + self.routing_key.as_str().encoded_size()
+            + self.super_stream.as_str().encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.routing_key.as_str().encode(writer)?;
+        self.super_stream.as_str().encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Decoder for SuperStreamRouteRequest {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, routing_key) = Option::decode(input)?;
+        let (input, super_stream) = Option::decode(input)?;
+
+        Ok((
+            input,
+            SuperStreamRouteRequest {
+                correlation_id,
+                routing_key: routing_key.unwrap(),
+                super_stream: super_stream.unwrap(),
+            },
+        ))
+    }
+}
+
+impl Command for SuperStreamRouteRequest {
+    fn key(&self) -> u16 {
+        COMMAND_ROUTE
+    }
+}
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct SuperStreamRouteResponse {
+    pub(crate) correlation_id: u32,
+    response_code: ResponseCode,
+    pub streams: Vec<String>,
+}
+
+impl SuperStreamRouteResponse {
+    pub fn new(correlation_id: u32, streams: Vec<String>, response_code: ResponseCode) -> Self {
+        Self {
+            correlation_id,
+            response_code,
+            streams,
+        }
+    }
+    pub fn is_ok(&self) -> bool {
+        self.response_code == ResponseCode::Ok
+    }
+}
+
+impl Encoder for SuperStreamRouteResponse {
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.response_code.encode(writer)?;
+        self.streams.encode(writer)?;
+        Ok(())
+    }
+
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size()
+            + self.streams.encoded_size()
+            + self.response_code.encoded_size()
+    }
+}
+
+impl Decoder for SuperStreamRouteResponse {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, response_code) = ResponseCode::decode(input)?;
+        let (input, streams) = Vec::decode(input)?;
+
+        Ok((
+            input,
+            SuperStreamRouteResponse {
+                correlation_id,
+                response_code,
+                streams,
+            },
+        ))
+    }
+}
+
+impl FromResponse for SuperStreamRouteResponse {
+    fn from_response(response: crate::Response) -> Option<Self> {
+        match response.kind {
+            crate::ResponseKind::SuperStreamRoute(route) => Some(route),
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use crate::commands::tests::command_encode_decode_test;
+
+    use super::SuperStreamRouteRequest;
+    use super::SuperStreamRouteResponse;
+
+    #[test]
+    fn super_stream_route_request_test() {
+        command_encode_decode_test::<SuperStreamRouteRequest>();
+    }
+
+    #[test]
+    fn super_stream_route_response_test() {
+        command_encode_decode_test::<SuperStreamRouteResponse>();
+    }
+}