Superstream: Create/Delete superstream and Partition and route commands (#230)

* WIP: implementing create and delete superstream commands

* adding superstream error-case test

* implementing route and partition commands

* refactoring tests

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

@@ -0,0 +1,100 @@
+use std::collections::HashMap;
+use std::io::Write;
+
+#[cfg(test)]
+use fake::Fake;
+
+use crate::{
+    codec::{Decoder, Encoder},
+    error::{DecodeError, EncodeError},
+    protocol::commands::COMMAND_CREATE_SUPER_STREAM,
+};
+
+use super::Command;
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct CreateSuperStreamCommand {
+    correlation_id: u32,
+    super_stream_name: String,
+    partitions: Vec<String>,
+    binding_keys: Vec<String>,
+    args: HashMap<String, String>,
+}
+
+impl CreateSuperStreamCommand {
+    pub fn new(
+        correlation_id: u32,
+        super_stream_name: String,
+        partitions: Vec<String>,
+        binding_keys: Vec<String>,
+        args: HashMap<String, String>,
+    ) -> Self {
+        Self {
+            correlation_id,
+            super_stream_name,
+            partitions,
+            binding_keys,
+            args,
+        }
+    }
+}
+
+impl Encoder for CreateSuperStreamCommand {
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size()
+            + self.super_stream_name.as_str().encoded_size()
+            + self.partitions.encoded_size()
+            + self.binding_keys.encoded_size()
+            + self.args.encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.super_stream_name.as_str().encode(writer)?;
+        self.partitions.encode(writer)?;
+        self.binding_keys.encode(writer)?;
+        self.args.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Decoder for CreateSuperStreamCommand {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, super_stream_name) = Option::decode(input)?;
+        let (input, partitions) = <Vec<String>>::decode(input)?;
+        let (input, binding_keys) = <Vec<String>>::decode(input)?;
+        let (input, args) = HashMap::decode(input)?;
+
+        Ok((
+            input,
+            CreateSuperStreamCommand {
+                correlation_id,
+                super_stream_name: super_stream_name.unwrap(),
+                partitions,
+                binding_keys,
+                args,
+            },
+        ))
+    }
+}
+
+impl Command for CreateSuperStreamCommand {
+    fn key(&self) -> u16 {
+        COMMAND_CREATE_SUPER_STREAM
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use crate::commands::tests::command_encode_decode_test;
+
+    use super::CreateSuperStreamCommand;
+
+    #[test]
+    fn create_super_stream_request_test() {
+        command_encode_decode_test::<CreateSuperStreamCommand>();
+    }
+}

protocol/src/commands/delete_super_stream.rs

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

protocol/src/commands/mod.rs

@@ -2,10 +2,12 @@ use crate::protocol::version::PROTOCOL_VERSION;
 
 pub mod close;
 pub mod create_stream;
+pub mod create_super_stream;
 pub mod credit;
 pub mod declare_publisher;
 pub mod delete;
 pub mod delete_publisher;
+pub mod delete_super_stream;
 pub mod deliver;
 pub mod exchange_command_versions;
 pub mod generic;
@@ -23,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,144 @@
+use std::io::Write;
+
+#[cfg(test)]
+use fake::Fake;
+
+use super::Command;
+use crate::{
+    codec::{Decoder, Encoder},
+    error::{DecodeError, EncodeError},
+    protocol::commands::COMMAND_PARTITIONS,
+    FromResponse, ResponseCode,
+};
+
+#[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>();
+    }
+}