Support for publish/consuming messages

Co-authored-by: Michael Davis <[email protected]>

Author: mkuratczyk

src/blocking.rs

@@ -1,7 +1,7 @@
 use crate::{
     commons::{BindingDestinationType, UserLimitTarget, VirtualHostLimitTarget},
     requests::{
-        EnforcedLimitParams, ExchangeParams, Permissions, PolicyParams, QueueParams,
+        self, EnforcedLimitParams, ExchangeParams, Permissions, PolicyParams, QueueParams,
         RuntimeParameterDefinition, UserParams, VirtualHostParams, XArguments,
     },
     responses::{self, BindingInfo},
@@ -1051,6 +1051,65 @@ impl<'a> Client<'a> {
         ))
     }
 
+    //
+    // Publish and consume messages
+    //
+    pub fn publish_message(
+        &self,
+        vhost: &str,
+        exchange: &str,
+        routing_key: &str,
+        payload: &str,
+        properties: requests::MessageProperties,
+    ) -> Result<responses::MessageRouted> {
+        let url = format!(
+            "exchanges/{}/{}/publish",
+            self.percent_encode(vhost),
+            self.percent_encode(exchange)
+        );
+
+        let body = serde_json::json!({
+          "routing_key": routing_key,
+          "payload": payload,
+          "payload_encoding": "string",
+          "properties": properties,
+        });
+
+        let response = self.http_post(&url, &body)?;
+
+        let response2 = self.ok_or_status_code_error(response)?;
+        response2
+            .json::<responses::MessageRouted>()
+            .map_err(Error::from)
+    }
+
+    pub fn get_messages(
+        &self,
+        vhost: &str,
+        queue: &str,
+        count: i32,
+        ack_mode: &str,
+    ) -> Result<Vec<responses::GetMessage>> {
+        let url = format!(
+            "queues/{}/{}/get",
+            self.percent_encode(vhost),
+            self.percent_encode(queue)
+        );
+
+        let body = serde_json::json!({
+          "count": count,
+          "ackmode": ack_mode,
+          "encoding": "auto"
+        });
+
+        let response = self.http_post(&url, &body)?;
+
+        let response2 = self.ok_or_status_code_error(response)?;
+        response2
+            .json::<Vec<responses::GetMessage>>()
+            .map_err(Error::from)
+    }
+
     //
     // Implementation
     //

src/requests.rs

@@ -269,3 +269,5 @@ pub struct Permissions<'a> {
     pub read: &'a str,
     pub write: &'a str,
 }
+
+pub type MessageProperties = Map<String, Value>;

src/responses.rs

@@ -66,7 +66,8 @@ impl fmt::Display for XArguments {
     }
 }
 
-#[derive(Debug, Deserialize, Clone)]
+#[derive(Debug, Deserialize, Clone, Default)]
+#[serde(transparent)]
 pub struct RuntimeParameterValue(pub Map<String, serde_json::Value>);
 impl fmt::Display for RuntimeParameterValue {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -524,43 +525,6 @@ pub struct RuntimeParameter {
     pub value: RuntimeParameterValue,
 }
 
-fn deserialize_runtime_parameter_value<'de, D>(
-    deserializer: D,
-) -> Result<RuntimeParameterValue, D::Error>
-where
-    D: serde::Deserializer<'de>,
-{
-    struct RuntimeParameterValueVisitor;
-
-    impl<'de> Visitor<'de> for RuntimeParameterValueVisitor {
-        type Value = RuntimeParameterValue;
-
-        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
-            formatter.write_str("a runtime parameter")
-        }
-
-        fn visit_seq<A>(self, _seq: A) -> Result<Self::Value, A::Error>
-        where
-            A: serde::de::SeqAccess<'de>,
-        {
-            // Always deserialize the value as a map, even if the server
-            // sends a sequence.
-            Ok(RuntimeParameterValue(Map::new()))
-        }
-
-        fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
-        where
-            A: MapAccess<'de>,
-        {
-            let deserializer = serde::de::value::MapAccessDeserializer::new(map);
-            let m = Deserialize::deserialize(deserializer)?;
-            Ok(RuntimeParameterValue(m))
-        }
-    }
-
-    deserializer.deserialize_any(RuntimeParameterValueVisitor)
-}
-
 #[derive(Debug, Deserialize, Clone)]
 #[allow(dead_code)]
 pub struct ClusterIdentity {
@@ -639,6 +603,104 @@ pub struct QuorumEndangeredQueue {
     pub queue_type: String,
 }
 
+#[derive(Debug, Deserialize, Clone, Eq, PartialEq)]
+#[cfg_attr(feature = "tabled", derive(Tabled))]
+#[allow(dead_code)]
+pub struct GetMessage {
+    pub payload_bytes: i32,
+    pub redelivered: bool,
+    pub exchange: String,
+    pub routing_key: String,
+    pub message_count: i32,
+    #[serde(deserialize_with = "deserialize_message_properties")]
+    pub properties: MessageProperties,
+    pub payload: String,
+    pub payload_encoding: String,
+}
+
+#[derive(Debug, Deserialize, Clone, Eq, PartialEq)]
+pub struct MessageRouted {
+    pub routed: bool,
+}
+
+impl fmt::Display for MessageRouted {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self.routed {
+            true => write!(f, "Message published and routed successfully"),
+            false => write!(f, "Message published but NOT routed"),
+        }
+    }
+}
+
+#[derive(Debug, Deserialize, Clone, Eq, PartialEq, Default)]
+#[serde(transparent)]
+pub struct MessageProperties(pub Map<String, serde_json::Value>);
+
+impl fmt::Display for MessageProperties {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        for (k, v) in &self.0 {
+            writeln!(f, "{}: {}", k, v)?;
+        }
+
+        Ok(())
+    }
+}
+
 fn undefined() -> String {
     "?".to_string()
 }
+
+fn deserialize_map_or_seq<'de, T, D>(deserializer: D) -> Result<T, D::Error>
+where
+    T: Default + serde::Deserialize<'de>,
+    D: serde::Deserializer<'de>,
+{
+    struct MapVisitor<T> {
+        default: T,
+    }
+
+    impl<'de, T: serde::Deserialize<'de>> Visitor<'de> for MapVisitor<T> {
+        type Value = T;
+
+        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+            formatter.write_str("map")
+        }
+
+        fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+        where
+            A: MapAccess<'de>,
+        {
+            let deserializer = serde::de::value::MapAccessDeserializer::new(map);
+            let m = Deserialize::deserialize(deserializer)?;
+            Ok(m)
+        }
+
+        fn visit_seq<A>(self, _seq: A) -> Result<Self::Value, A::Error>
+        where
+            A: serde::de::SeqAccess<'de>,
+        {
+            // Treat a sequence as the default for the type.
+            Ok(self.default)
+        }
+    }
+
+    deserializer.deserialize_any(MapVisitor {
+        default: T::default(),
+    })
+}
+
+fn deserialize_message_properties<'de, D>(deserializer: D) -> Result<MessageProperties, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    deserialize_map_or_seq::<MessageProperties, D>(deserializer)
+}
+
+fn deserialize_runtime_parameter_value<'de, D>(
+    deserializer: D,
+) -> Result<RuntimeParameterValue, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    deserialize_map_or_seq::<RuntimeParameterValue, D>(deserializer)
+}

tests/message_tests.rs

@@ -0,0 +1,78 @@
+use rabbitmq_http_client::{
+    blocking::Client,
+    requests::{self, QueueParams},
+    responses::{GetMessage, MessageProperties, MessageRouted},
+};
+use serde_json::{json, Map, Value};
+
+mod common;
+use crate::common::{endpoint, PASSWORD, USERNAME};
+
+#[test]
+fn test_publish_and_get() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint).with_basic_auth_credentials(USERNAME, PASSWORD);
+    let vhost = "/";
+    let queue = "rust.tests.cq.publish_and_get";
+
+    let _ = rc.delete_queue(vhost, queue);
+
+    let params = QueueParams::new_durable_classic_queue(queue, None);
+    let result2 = rc.declare_queue(vhost, &params);
+    assert!(result2.is_ok(), "declare_queue returned {:?}", result2);
+
+    let result3 = rc.publish_message(
+        vhost,
+        "",
+        queue,
+        "rust test 1",
+        requests::MessageProperties::default(),
+    );
+    assert!(result3.is_ok(), "get_messages returned {:?}", result3);
+    assert_eq!(result3.unwrap(), MessageRouted { routed: true });
+
+    let mut props = Map::<String, Value>::new();
+    props.insert(String::from("timestamp"), json!(123456789));
+    let result4 = rc.publish_message(vhost, "", queue, "rust test 2", props.clone());
+    assert!(result4.is_ok(), "get_messages returned {:?}", result4);
+    assert_eq!(result4.unwrap(), MessageRouted { routed: true });
+
+    let result5 = rc.get_messages(vhost, queue, 1, "ack_requeue_false");
+    assert!(result5.is_ok(), "get_messages returned {:?}", result5);
+
+    let result6 = result5.unwrap();
+    assert_eq!(
+        result6,
+        [GetMessage {
+            payload_bytes: 11,
+            redelivered: false,
+            exchange: "".to_owned(),
+            routing_key: "rust.tests.cq.publish_and_get".to_owned(),
+            message_count: 1,
+            properties: MessageProperties::default(),
+            payload: "rust test 1".to_owned(),
+            payload_encoding: "string".to_owned()
+        }]
+    );
+
+    let result7 = rc.get_messages(vhost, queue, 1, "ack_requeue_false");
+    assert!(result7.is_ok(), "get_messages returned {:?}", result7);
+
+    let props = rabbitmq_http_client::responses::MessageProperties(props);
+    let result8 = result7.unwrap();
+    assert_eq!(
+        result8,
+        [GetMessage {
+            payload_bytes: 11,
+            redelivered: false,
+            exchange: "".to_owned(),
+            routing_key: "rust.tests.cq.publish_and_get".to_owned(),
+            message_count: 0,
+            properties: props,
+            payload: "rust test 2".to_owned(),
+            payload_encoding: "string".to_owned()
+        }]
+    );
+
+    rc.delete_queue(vhost, queue).unwrap();
+}