Introduce a function for declaring an AMQP 1.0 shovel

Author: michaelklishin

CHANGELOG.md

@@ -4,6 +4,11 @@
 
 ### Enhancements
 
+ * `Client#declare_amqp10_shovel` is a new function that declares a dynamic shovel
+   where both source and destination use AMQP 1.0
+
+   
+
  * Both `Amqp091ShovelSourceParams` and `Amqp091ShovelDestinationParams` now support a new boolean option, `predeclared`,
    that enables [either or both sides to rely on a pre-declared topology](https://www.rabbitmq.com/docs/shovel-dynamic#predeclared-topology)
 

src/api.rs

@@ -24,7 +24,9 @@ use std::fmt;
 
 use crate::error::Error;
 use crate::error::Error::{ClientErrorResponse, NotFound, ServerErrorResponse};
-use crate::requests::{Amqp091ShovelParams, EmptyPayload, StreamParams, SHOVEL_COMPONENT};
+use crate::requests::{
+    Amqp091ShovelParams, Amqp10ShovelParams, EmptyPayload, StreamParams, SHOVEL_COMPONENT,
+};
 use crate::responses::{
     DeprecatedFeatureList, FeatureFlag, FeatureFlagList, FeatureFlagStability, FeatureFlagState,
     GetMessage, OAuthConfiguration, SchemaDefinitionSyncStatus, VirtualHostDefinitionSet,
@@ -1392,6 +1394,25 @@ where
         Ok(())
     }
 
+    pub async fn declare_amqp10_shovel(&self, params: Amqp10ShovelParams<'_>) -> Result<()> {
+        let runtime_param = RuntimeParameterDefinition::from(params);
+
+        let _response = self
+            .http_put(
+                path!(
+                    "parameters",
+                    SHOVEL_COMPONENT,
+                    runtime_param.vhost,
+                    runtime_param.name
+                ),
+                &runtime_param,
+                None,
+                None,
+            )
+            .await?;
+        Ok(())
+    }
+
     pub async fn delete_shovel(&self, vhost: &str, name: &str, idempotently: bool) -> Result<()> {
         let excludes = if idempotently {
             Some(StatusCode::NOT_FOUND)

src/blocking_api.rs

@@ -15,7 +15,9 @@
 
 use crate::error::Error;
 use crate::error::Error::{ClientErrorResponse, NotFound, ServerErrorResponse};
-use crate::requests::{Amqp091ShovelParams, EmptyPayload, StreamParams, SHOVEL_COMPONENT};
+use crate::requests::{
+    Amqp091ShovelParams, Amqp10ShovelParams, EmptyPayload, StreamParams, SHOVEL_COMPONENT,
+};
 use crate::responses::{
     DeprecatedFeatureList, FeatureFlag, FeatureFlagList, FeatureFlagStability, FeatureFlagState,
     GetMessage, OAuthConfiguration, VirtualHostDefinitionSet, WarmStandbyReplicationStatus,
@@ -1223,6 +1225,23 @@ where
         Ok(())
     }
 
+    pub fn declare_amqp10_shovel(&self, params: Amqp10ShovelParams<'_>) -> Result<()> {
+        let runtime_param = RuntimeParameterDefinition::from(params);
+
+        let _response = self.http_put(
+            path!(
+                "parameters",
+                SHOVEL_COMPONENT,
+                runtime_param.vhost,
+                runtime_param.name
+            ),
+            &runtime_param,
+            None,
+            None,
+        )?;
+        Ok(())
+    }
+
     pub fn delete_shovel(&self, vhost: &str, name: &str, idempotently: bool) -> Result<()> {
         let excludes = if idempotently {
             Some(StatusCode::NOT_FOUND)

src/requests.rs

@@ -406,7 +406,7 @@ pub struct Permissions<'a> {
 
 pub(crate) const SHOVEL_COMPONENT: &str = "shovel";
 
-/// Represents a dynamic shovel definition.
+/// Represents a dynamic AMQP 0-9-1 shovel definition.
 #[derive(Serialize)]
 pub struct Amqp091ShovelParams<'a> {
     pub name: &'a str,
@@ -423,6 +423,9 @@ impl<'a> From<Amqp091ShovelParams<'a>> for RuntimeParameterDefinition<'a> {
     fn from(params: Amqp091ShovelParams<'a>) -> Self {
         let mut value = Map::new();
 
+        value.insert("src-protocol".to_owned(), json!("amqp091"));
+        value.insert("dest-protocol".to_owned(), json!("amqp091"));
+
         value.insert("src-uri".to_owned(), json!(params.source.source_uri));
         if let Some(sq) = params.source.source_queue {
             value.insert("src-queue".to_owned(), json!(sq));
@@ -614,6 +617,85 @@ impl<'a> Amqp091ShovelDestinationParams<'a> {
     }
 }
 
+/// Represents a dynamic shovel definition.
+#[derive(Serialize)]
+pub struct Amqp10ShovelParams<'a> {
+    pub name: &'a str,
+    pub vhost: &'a str,
+
+    pub acknowledgement_mode: ShovelAcknowledgementMode,
+    pub reconnect_delay: Option<u16>,
+
+    pub source: Amqp10ShovelSourceParams<'a>,
+    pub destination: Amqp10ShovelDestinationParams<'a>,
+}
+
+#[derive(Serialize)]
+pub struct Amqp10ShovelSourceParams<'a> {
+    pub source_uri: &'a str,
+    pub source_address: &'a str,
+}
+
+impl<'a> Amqp10ShovelSourceParams<'a> {
+    pub fn new(uri: &'a str, address: &'a str) -> Self {
+        Self {
+            source_uri: uri,
+            source_address: address,
+        }
+    }
+}
+
+impl<'a> From<Amqp10ShovelParams<'a>> for RuntimeParameterDefinition<'a> {
+    fn from(params: Amqp10ShovelParams<'a>) -> Self {
+        let mut value = Map::new();
+
+        value.insert("src-protocol".to_owned(), json!("amqp10"));
+        value.insert("dest-protocol".to_owned(), json!("amqp10"));
+
+        value.insert("src-uri".to_owned(), json!(params.source.source_uri));
+        value.insert(
+            "src-address".to_owned(),
+            json!(params.source.source_address),
+        );
+
+        value.insert(
+            "dest-uri".to_owned(),
+            json!(params.destination.destination_uri),
+        );
+        value.insert(
+            "dest-address".to_owned(),
+            json!(params.destination.destination_address),
+        );
+
+        value.insert("ack-mode".to_owned(), json!(params.acknowledgement_mode));
+        if let Some(val) = params.reconnect_delay {
+            value.insert("reconnect-delay".to_owned(), json!(val));
+        }
+
+        Self {
+            name: params.name,
+            vhost: params.vhost,
+            component: SHOVEL_COMPONENT,
+            value,
+        }
+    }
+}
+
+#[derive(Serialize)]
+pub struct Amqp10ShovelDestinationParams<'a> {
+    pub destination_uri: &'a str,
+    pub destination_address: &'a str,
+}
+
+impl<'a> Amqp10ShovelDestinationParams<'a> {
+    pub fn new(uri: &'a str, address: &'a str) -> Self {
+        Self {
+            destination_uri: uri,
+            destination_address: address,
+        }
+    }
+}
+
 pub type MessageProperties = Map<String, Value>;
 
 #[derive(Serialize, Default)]

src/responses.rs

@@ -1612,6 +1612,13 @@ pub struct Shovel {
     #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
     pub destination: Option<String>,
 
+    #[serde(rename = "src_address")]
+    #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
+    pub source_address: Option<String>,
+    #[serde(rename = "dest_address")]
+    #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
+    pub destination_address: Option<String>,
+
     #[serde(rename = "src_protocol")]
     #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
     pub source_protocol: Option<MessagingProtocol>,

tests/async_dynamic_shovel_tests.rs

@@ -13,7 +13,8 @@
 // limitations under the License.
 use rabbitmq_http_client::commons::ShovelAcknowledgementMode;
 use rabbitmq_http_client::requests::{
-    Amqp091ShovelDestinationParams, Amqp091ShovelParams, Amqp091ShovelSourceParams, QueueParams,
+    Amqp091ShovelDestinationParams, Amqp091ShovelParams, Amqp091ShovelSourceParams,
+    Amqp10ShovelDestinationParams, Amqp10ShovelParams, Amqp10ShovelSourceParams, QueueParams,
 };
 use rabbitmq_http_client::{api::Client, requests::VirtualHostParams};
 
@@ -56,6 +57,48 @@ async fn test_async_declare_a_dynamic_amqp091_shovel() {
     let _ = rc.delete_vhost(vh_params.name, false).await;
 }
 
+#[tokio::test]
+async fn test_async_declare_a_dynamic_amqp10_shovel() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint, USERNAME, PASSWORD);
+
+    let vh = "rust.http.api.async.test_async_declare_a_dynamic_amqp10_shovel";
+    let sh = "test_async_declare_a_dynamic_amqp10_shovel";
+
+    let vh_params = VirtualHostParams::named(vh);
+    let result1 = rc.create_vhost(&vh_params).await;
+    assert!(result1.is_ok());
+
+    // note: 4.1.0 will use a different addressing scheme,
+    //       see https://www.rabbitmq.com/docs/next/amqp#address-v2
+    let src_queue = format!("{}.src.q", sh);
+    let src_address = format!("/queue/{}.src.q", sh);
+    let dest_queue = format!("{}.dest.q", sh);
+    let dest_address = format!("/queue/{}.dest.q", sh);
+
+    let src_params = QueueParams::new_durable_classic_queue(&src_queue, None);
+    let result2 = rc.declare_queue(vh, &src_params).await;
+    assert!(result2.is_ok());
+
+    let dest_params = QueueParams::new_durable_classic_queue(&dest_queue, None);
+    let result3 = rc.declare_queue(vh, &dest_params).await;
+    assert!(result3.is_ok());
+
+    let amqp_endpoint = amqp_endpoint_with_vhost(&vh);
+    let shovel_params = Amqp10ShovelParams {
+        vhost: &vh,
+        name: sh,
+        acknowledgement_mode: ShovelAcknowledgementMode::WhenConfirmed,
+        reconnect_delay: Some(5),
+        source: Amqp10ShovelSourceParams::new(&amqp_endpoint, &src_address),
+        destination: Amqp10ShovelDestinationParams::new(&amqp_endpoint, &dest_address),
+    };
+    let result4 = rc.declare_amqp10_shovel(shovel_params).await;
+    assert!(result4.is_ok());
+
+    let _ = rc.delete_vhost(vh_params.name, false).await;
+}
+
 #[tokio::test]
 async fn test_async_declare_a_dynamic_amqp091_shovel_with_predeclared_source_topology() {
     let endpoint = endpoint();

tests/blocking_dynamic_shovel_tests.rs

@@ -13,7 +13,8 @@
 // limitations under the License.
 use rabbitmq_http_client::commons::ShovelAcknowledgementMode;
 use rabbitmq_http_client::requests::{
-    Amqp091ShovelDestinationParams, Amqp091ShovelParams, Amqp091ShovelSourceParams, QueueParams,
+    Amqp091ShovelDestinationParams, Amqp091ShovelParams, Amqp091ShovelSourceParams,
+    Amqp10ShovelDestinationParams, Amqp10ShovelParams, Amqp10ShovelSourceParams, QueueParams,
 };
 use rabbitmq_http_client::{blocking_api::Client, requests::VirtualHostParams};
 
@@ -56,6 +57,48 @@ fn test_blocking_declare_a_dynamic_amqp091_shovel() {
     let _ = rc.delete_vhost(vh_params.name, false);
 }
 
+#[test]
+fn test_blocking_declare_a_dynamic_amqp10_shovel() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint, USERNAME, PASSWORD);
+
+    let vh = "rust.http.api.blocking.test_blocking_declare_a_dynamic_amqp10_shovel";
+    let sh = "test_async_declare_a_dynamic_amqp10_shovel";
+
+    let vh_params = VirtualHostParams::named(vh);
+    let result1 = rc.create_vhost(&vh_params);
+    assert!(result1.is_ok());
+
+    // note: 4.1.0 will use a different addressing scheme,
+    //       see https://www.rabbitmq.com/docs/next/amqp#address-v2
+    let src_queue = format!("{}.src.q", sh);
+    let src_address = format!("/queue/{}.src.q", sh);
+    let dest_queue = format!("{}.dest.q", sh);
+    let dest_address = format!("/queue/{}.dest.q", sh);
+
+    let src_params = QueueParams::new_durable_classic_queue(&src_queue, None);
+    let result2 = rc.declare_queue(vh, &src_params);
+    assert!(result2.is_ok());
+
+    let dest_params = QueueParams::new_durable_classic_queue(&dest_queue, None);
+    let result3 = rc.declare_queue(vh, &dest_params);
+    assert!(result3.is_ok());
+
+    let amqp_endpoint = amqp_endpoint_with_vhost(&vh);
+    let shovel_params = Amqp10ShovelParams {
+        vhost: &vh,
+        name: sh,
+        acknowledgement_mode: ShovelAcknowledgementMode::WhenConfirmed,
+        reconnect_delay: Some(5),
+        source: Amqp10ShovelSourceParams::new(&amqp_endpoint, &src_address),
+        destination: Amqp10ShovelDestinationParams::new(&amqp_endpoint, &dest_address),
+    };
+    let result4 = rc.declare_amqp10_shovel(shovel_params);
+    assert!(result4.is_ok());
+
+    let _ = rc.delete_vhost(vh_params.name, false);
+}
+
 #[test]
 fn test_blocking_declare_a_dynamic_amqp091_shovel_with_predeclared_source_topology() {
     let endpoint = endpoint();