Filtering supported (#225)

* Filtering supported

Author: JiaYingZhang

examples/filtering.rs

@@ -0,0 +1,84 @@
+use futures::StreamExt;
+use rabbitmq_stream_client::types::{Message, OffsetSpecification};
+use rabbitmq_stream_client::{Environment, FilterConfiguration};
+use tracing::info;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let environment = Environment::builder()
+        .host("localhost")
+        .port(5552)
+        .build()
+        .await?;
+
+    let message_count = 10;
+    environment.stream_creator().create("test").await?;
+
+    let mut producer = environment
+        .producer()
+        .name("test_producer")
+        .filter_value_extractor(|message| {
+            String::from_utf8(message.data().unwrap().to_vec()).unwrap()
+        })
+        .build("test")
+        .await?;
+
+    // publish filtering message
+    for i in 0..message_count {
+        producer
+            .send_with_confirm(Message::builder().body(i.to_string()).build())
+            .await?;
+    }
+
+    producer.close().await?;
+
+    // publish filtering message
+    let mut producer = environment
+        .producer()
+        .name("test_producer")
+        .build("test")
+        .await?;
+
+    // publish unset filter value
+    for i in 0..message_count {
+        producer
+            .send_with_confirm(Message::builder().body(i.to_string()).build())
+            .await?;
+    }
+
+    producer.close().await?;
+
+    // filter configuration: https://www.rabbitmq.com/blog/2023/10/16/stream-filtering
+    let filter_configuration =
+        FilterConfiguration::new(vec!["1".to_string()], false).post_filter(|message| {
+            String::from_utf8(message.data().unwrap().to_vec()).unwrap_or("".to_string())
+                == "1".to_string()
+        });
+    // let filter_configuration = FilterConfiguration::new(vec!["1".to_string()], true);
+
+    let mut consumer = environment
+        .consumer()
+        .offset(OffsetSpecification::First)
+        .filter_input(Some(filter_configuration))
+        .build("test")
+        .await
+        .unwrap();
+
+    let task = tokio::task::spawn(async move {
+        loop {
+            let delivery = consumer.next().await.unwrap().unwrap();
+            info!(
+                "Got message : {:?} with offset {}",
+                delivery
+                    .message()
+                    .data()
+                    .map(|data| String::from_utf8(data.to_vec())),
+                delivery.offset()
+            );
+        }
+    });
+    let _ = tokio::time::timeout(tokio::time::Duration::from_secs(3), task).await;
+
+    environment.delete_stream("test").await?;
+    Ok(())
+}

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>();
+    }
+}