Filtering supported

examples/raw_client.rs

@@ -39,6 +39,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 Message::builder()
                     .body(format!("message {}", i).as_bytes().to_vec())
                     .build(),
+                1,
             )
             .await
             .unwrap();

protocol/src/codec/decoder.rs

@@ -106,7 +106,18 @@ impl Decoder for PublishedMessage {
         let (input, publishing_id) = u64::decode(input)?;
         let (input, body) = read_vec::<u8>(input)?;
         let (_, message) = Message::decode(&body)?;
-        Ok((input, PublishedMessage::new(publishing_id, message)))
+        Ok((input, PublishedMessage::new(publishing_id, message, None)))
+    }
+
+    fn decode_version_2(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, publishing_id) = u64::decode(input)?;
+        let (input, body) = read_vec::<u8>(input)?;
+        let (input, filter_value) = <Option<String>>::decode(input)?;
+        let (_, message) = Message::decode(&body)?;
+        Ok((
+            input,
+            PublishedMessage::new(publishing_id, message, filter_value),
+        ))
     }
 }
 

protocol/src/codec/encoder.rs

@@ -109,6 +109,21 @@ impl Encoder for PublishedMessage {
         self.message.encode(writer)?;
         Ok(())
     }
+
+    fn encoded_size_version_2(&self) -> u32 {
+        self.publishing_id.encoded_size()
+            + self.filter_value.encoded_size()
+            + 4
+            + self.message.encoded_size()
+    }
+
+    fn encode_version_2(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.publishing_id.encode(writer)?;
+        self.filter_value.encode(writer)?;
+        self.message.encoded_size().encode(writer)?;
+        self.message.encode(writer)?;
+        Ok(())
+    }
 }
 
 impl Encoder for Vec<PublishedMessage> {
@@ -123,6 +138,20 @@ impl Encoder for Vec<PublishedMessage> {
         }
         Ok(())
     }
+
+    fn encoded_size_version_2(&self) -> u32 {
+        4 + self
+            .iter()
+            .fold(0, |acc, v| acc + v.encoded_size_version_2())
+    }
+
+    fn encode_version_2(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        writer.write_u32::<BigEndian>(self.len() as u32)?;
+        for x in self {
+            x.encode_version_2(writer)?;
+        }
+        Ok(())
+    }
 }
 
 impl Encoder for &str {

protocol/src/codec/mod.rs

@@ -8,11 +8,20 @@ pub mod encoder;
 pub trait Encoder {
     fn encoded_size(&self) -> u32;
     fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError>;
+    fn encoded_size_version_2(&self) -> u32 {
+        self.encoded_size()
+    }
+    fn encode_version_2(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.encode(writer)
+    }
 }
 
 pub trait Decoder
 where
     Self: Sized,
 {
     fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError>;
+    fn decode_version_2(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        Decoder::decode(input)
+    }
 }

protocol/src/commands/exchange_command_versions.rs

@@ -0,0 +1,221 @@
+use std::io::Write;
+
+use crate::{
+    codec::{decoder::read_vec, Decoder, Encoder},
+    error::{DecodeError, EncodeError},
+    protocol::commands::COMMAND_EXCHANGE_COMMAND_VERSIONS,
+    response::{FromResponse, ResponseCode},
+};
+
+use super::Command;
+use byteorder::{BigEndian, WriteBytesExt};
+
+#[cfg(test)]
+use fake::Fake;
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct ExchangeCommandVersion(u16, u16, u16);
+
+impl ExchangeCommandVersion {
+    pub fn new(key: u16, min_version: u16, max_version: u16) -> Self {
+        ExchangeCommandVersion(key, min_version, max_version)
+    }
+}
+
+impl Encoder for ExchangeCommandVersion {
+    fn encoded_size(&self) -> u32 {
+        self.0.encoded_size() + self.1.encoded_size() + self.2.encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.0.encode(writer)?;
+        self.1.encode(writer)?;
+        self.2.encode(writer)?;
+
+        Ok(())
+    }
+}
+
+impl Decoder for ExchangeCommandVersion {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, key) = u16::decode(input)?;
+        let (input, min_version) = u16::decode(input)?;
+        let (input, max_version) = u16::decode(input)?;
+        Ok((input, ExchangeCommandVersion(key, min_version, max_version)))
+    }
+}
+
+impl Encoder for Vec<ExchangeCommandVersion> {
+    fn encoded_size(&self) -> u32 {
+        4 + self.iter().fold(0, |acc, v| acc + v.encoded_size())
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        writer.write_u32::<BigEndian>(self.len() as u32)?;
+        for x in self {
+            x.encode(writer)?;
+        }
+        Ok(())
+    }
+}
+
+impl Decoder for Vec<ExchangeCommandVersion> {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, result) = read_vec(input)?;
+        Ok((input, result))
+    }
+}
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct ExchangeCommandVersionsRequest {
+    pub(crate) correlation_id: u32,
+    commands: Vec<ExchangeCommandVersion>,
+}
+
+impl ExchangeCommandVersionsRequest {
+    pub fn new(correlation_id: u32, commands: Vec<ExchangeCommandVersion>) -> Self {
+        Self {
+            correlation_id,
+            commands,
+        }
+    }
+}
+
+impl Encoder for ExchangeCommandVersionsRequest {
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size() + self.commands.encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.commands.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Command for ExchangeCommandVersionsRequest {
+    fn key(&self) -> u16 {
+        COMMAND_EXCHANGE_COMMAND_VERSIONS
+    }
+}
+
+impl Decoder for ExchangeCommandVersionsRequest {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, commands) = <Vec<ExchangeCommandVersion>>::decode(input)?;
+        Ok((
+            input,
+            ExchangeCommandVersionsRequest {
+                correlation_id,
+                commands,
+            },
+        ))
+    }
+}
+
+#[cfg_attr(test, derive(fake::Dummy))]
+#[derive(PartialEq, Eq, Debug)]
+pub struct ExchangeCommandVersionsResponse {
+    pub(crate) correlation_id: u32,
+    response_code: ResponseCode,
+    commands: Vec<ExchangeCommandVersion>,
+}
+
+impl ExchangeCommandVersionsResponse {
+    pub fn new(
+        correlation_id: u32,
+        response_code: ResponseCode,
+        commands: Vec<ExchangeCommandVersion>,
+    ) -> Self {
+        Self {
+            correlation_id,
+            response_code,
+            commands,
+        }
+    }
+
+    pub fn code(&self) -> &ResponseCode {
+        &self.response_code
+    }
+
+    pub fn is_ok(&self) -> bool {
+        self.response_code == ResponseCode::Ok
+    }
+
+    pub fn key_version(&self, key_command: u16) -> (u16, u16) {
+        for i in &self.commands {
+            match i {
+                ExchangeCommandVersion(match_key_command, min_version, max_version) => {
+                    if *match_key_command == key_command {
+                        return (*min_version, *max_version);
+                    }
+                }
+            }
+        }
+
+        (1, 1)
+    }
+}
+
+impl Encoder for ExchangeCommandVersionsResponse {
+    fn encoded_size(&self) -> u32 {
+        self.correlation_id.encoded_size()
+            + self.response_code.encoded_size()
+            + self.commands.encoded_size()
+    }
+
+    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
+        self.correlation_id.encode(writer)?;
+        self.response_code.encode(writer)?;
+        self.commands.encode(writer)?;
+        Ok(())
+    }
+}
+
+impl Decoder for ExchangeCommandVersionsResponse {
+    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
+        let (input, correlation_id) = u32::decode(input)?;
+        let (input, response_code) = ResponseCode::decode(input)?;
+        let (input, commands) = <Vec<ExchangeCommandVersion>>::decode(input)?;
+
+        Ok((
+            input,
+            ExchangeCommandVersionsResponse {
+                correlation_id,
+                response_code,
+                commands,
+            },
+        ))
+    }
+}
+
+impl FromResponse for ExchangeCommandVersionsResponse {
+    fn from_response(response: crate::Response) -> Option<Self> {
+        match response.kind {
+            crate::ResponseKind::ExchangeCommandVersions(exchange_command_versions) => {
+                Some(exchange_command_versions)
+            }
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use crate::commands::tests::command_encode_decode_test;
+
+    use super::{ExchangeCommandVersionsRequest, ExchangeCommandVersionsResponse};
+
+    #[test]
+    fn exchange_command_versions_request_test() {
+        command_encode_decode_test::<ExchangeCommandVersionsRequest>();
+    }
+
+    #[test]
+    fn exchange_command_versions_response_test() {
+        command_encode_decode_test::<ExchangeCommandVersionsResponse>();
+    }
+}

protocol/src/commands/mod.rs

@@ -1,10 +1,13 @@
+use crate::protocol::version::PROTOCOL_VERSION;
+
 pub mod close;
 pub mod create_stream;
 pub mod credit;
 pub mod declare_publisher;
 pub mod delete;
 pub mod delete_publisher;
 pub mod deliver;
+pub mod exchange_command_versions;
 pub mod generic;
 pub mod heart_beat;
 pub mod metadata;
@@ -25,6 +28,9 @@ pub mod unsubscribe;
 
 pub trait Command {
     fn key(&self) -> u16;
+    fn version(&self) -> u16 {
+        PROTOCOL_VERSION
+    }
 }
 
 #[cfg(test)]