Initial support for operations on Shovels

Author: michaelklishin

CHANGELOG.md

@@ -2,7 +2,16 @@
 
 ## v0.23.0  (in development)
 
-No (documented) changes yet.
+### Breaking Changes
+
+ * `RuntimeParameterDefinition#name`, `RuntimeParameterDefinition#vhost`, and `RuntimeParameterDefinition#component` types changed from `String` to `&str`
+
+### Enhancements
+
+ * `Client#declare_amqp091_shovel` is a new function that declares a dynamic shovel
+    where both source and destination use AMQP 0-9-1
+
+ * `Client#delete_shovel` is a new function for deleting shovels
 
 
 ## v0.22.0 (Feb 8, 2025)

src/api.rs

@@ -24,7 +24,7 @@ use std::fmt;
 
 use crate::error::Error;
 use crate::error::Error::{ClientErrorResponse, NotFound, ServerErrorResponse};
-use crate::requests::{EmptyPayload, StreamParams};
+use crate::requests::{Amqp091ShovelParams, EmptyPayload, StreamParams, SHOVEL_COMPONENT};
 use crate::responses::{
     DeprecatedFeatureList, FeatureFlag, FeatureFlagList, FeatureFlagStability, FeatureFlagState,
     GetMessage, OAuthConfiguration, SchemaDefinitionSyncStatus, VirtualHostDefinitionSet,
@@ -993,7 +993,10 @@ where
         Ok(response)
     }
 
-    pub async fn upsert_runtime_parameter(&self, param: &RuntimeParameterDefinition) -> Result<()> {
+    pub async fn upsert_runtime_parameter<'a>(
+        &self,
+        param: &'a RuntimeParameterDefinition<'a>,
+    ) -> Result<()> {
         let _response = self
             .http_put(
                 path!("parameters", param.component, param.vhost, param.name),
@@ -1364,12 +1367,43 @@ where
     // Shovels
     //
 
-    pub async fn list_shovels(&self) -> crate::blocking_api::Result<Vec<responses::Shovel>> {
+    pub async fn list_shovels(&self) -> Result<Vec<responses::Shovel>> {
         let response = self.http_get("shovels", None, None).await?;
         let response = response.json().await?;
         Ok(response)
     }
 
+    pub async fn declare_amqp091_shovel(&self, params: Amqp091ShovelParams<'_>) -> 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)
+        } else {
+            None
+        };
+        let _response = self
+            .http_delete(path!("shovels", "vhost", vhost, name), excludes, None)
+            .await?;
+        Ok(())
+    }
+
     //
     // Publish and consume messages
     //

src/blocking_api.rs

@@ -15,7 +15,7 @@
 
 use crate::error::Error;
 use crate::error::Error::{ClientErrorResponse, NotFound, ServerErrorResponse};
-use crate::requests::{EmptyPayload, StreamParams};
+use crate::requests::{Amqp091ShovelParams, EmptyPayload, StreamParams, SHOVEL_COMPONENT};
 use crate::responses::{
     DeprecatedFeatureList, FeatureFlag, FeatureFlagList, FeatureFlagStability, FeatureFlagState,
     GetMessage, OAuthConfiguration, VirtualHostDefinitionSet, WarmStandbyReplicationStatus,
@@ -879,7 +879,10 @@ where
         Ok(response)
     }
 
-    pub fn upsert_runtime_parameter(&self, param: &RuntimeParameterDefinition) -> Result<()> {
+    pub fn upsert_runtime_parameter<'a>(
+        &self,
+        param: &'a RuntimeParameterDefinition<'a>,
+    ) -> Result<()> {
         let _response = self.http_put(
             path!("parameters", param.component, param.vhost, param.name),
             &param,
@@ -1203,6 +1206,33 @@ where
         Ok(response)
     }
 
+    pub fn declare_amqp091_shovel(&self, params: Amqp091ShovelParams<'_>) -> 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)
+        } else {
+            None
+        };
+        let _response = self.http_delete(path!("shovels", "vhost", vhost, name), excludes, None)?;
+        Ok(())
+    }
+
     //
     // Publish and consume messages
     //

src/commons.rs

@@ -577,3 +577,36 @@ impl From<UserLimitTarget> for String {
         value.as_ref().to_string()
     }
 }
+
+#[derive(Default, Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
+pub enum ShovelAcknowledgementMode {
+    #[serde(rename = "no-ack")]
+    Immediate,
+    #[serde(rename = "on-publish")]
+    WhenPublished,
+    #[default]
+    #[serde(rename = "on-confirm")]
+    WhenConfirmed,
+}
+
+impl From<&str> for ShovelAcknowledgementMode {
+    fn from(value: &str) -> Self {
+        match value {
+            "no-ack" => ShovelAcknowledgementMode::Immediate,
+            "on-publish" => ShovelAcknowledgementMode::WhenPublished,
+            "on-confirm" => ShovelAcknowledgementMode::WhenConfirmed,
+            _ => ShovelAcknowledgementMode::default(),
+        }
+    }
+}
+
+impl From<String> for ShovelAcknowledgementMode {
+    fn from(value: String) -> Self {
+        match value.as_str() {
+            "no-ack" => ShovelAcknowledgementMode::Immediate,
+            "on-publish" => ShovelAcknowledgementMode::WhenPublished,
+            "on-confirm" => ShovelAcknowledgementMode::WhenConfirmed,
+            _ => ShovelAcknowledgementMode::default(),
+        }
+    }
+}

src/requests.rs

@@ -11,7 +11,7 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-use crate::commons::{ExchangeType, PolicyTarget, QueueType};
+use crate::commons::{ExchangeType, PolicyTarget, QueueType, ShovelAcknowledgementMode};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Map, Value};
 
@@ -373,10 +373,10 @@ pub type RuntimeParameterValue = Map<String, Value>;
 
 /// Represents a [runtime parameter](https://rabbitmq.com/docs/parameters/).
 #[derive(Serialize, Deserialize)]
-pub struct RuntimeParameterDefinition {
-    pub name: String,
-    pub vhost: String,
-    pub component: String,
+pub struct RuntimeParameterDefinition<'a> {
+    pub name: &'a str,
+    pub vhost: &'a str,
+    pub component: &'a str,
     pub value: RuntimeParameterValue,
 }
 
@@ -404,6 +404,136 @@ pub struct Permissions<'a> {
     pub write: &'a str,
 }
 
+pub(crate) const SHOVEL_COMPONENT: &str = "shovel";
+
+/// Represents a dynamic shovel definition.
+#[derive(Serialize)]
+pub struct Amqp091ShovelParams<'a> {
+    pub name: &'a str,
+    pub vhost: &'a str,
+
+    pub acknowledgement_mode: ShovelAcknowledgementMode,
+    pub reconnect_delay: Option<u16>,
+
+    pub source: Amqp091ShovelSourceParams<'a>,
+    pub destination: Amqp091ShovelDestinationParams<'a>,
+}
+
+impl<'a> From<Amqp091ShovelParams<'a>> for RuntimeParameterDefinition<'a> {
+    fn from(params: Amqp091ShovelParams<'a>) -> Self {
+        let mut value = Map::new();
+
+        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));
+        }
+        if let Some(sx) = params.source.source_exchange {
+            value.insert("src-exchange".to_owned(), json!(sx));
+        }
+        if let Some(sxrk) = params.source.source_exchange_routing_key {
+            value.insert("src-exchange-key".to_owned(), json!(sxrk));
+        }
+
+        value.insert(
+            "dest-uri".to_owned(),
+            json!(params.destination.destination_uri),
+        );
+        value.insert("ack-mode".to_owned(), json!(params.acknowledgement_mode));
+
+        if let Some(dq) = params.destination.destination_queue {
+            value.insert("dest-queue".to_owned(), json!(dq));
+        }
+        if let Some(dx) = params.destination.destination_exchange {
+            value.insert("dest-exchange".to_owned(), json!(dx));
+        }
+        if let Some(dxrk) = params.destination.destination_exchange_routing_key {
+            value.insert("dest-exchange-key".to_owned(), json!(dxrk));
+        }
+
+        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 Amqp091ShovelSourceParams<'a> {
+    pub source_uri: &'a str,
+
+    pub source_queue: Option<&'a str>,
+
+    pub source_exchange: Option<&'a str>,
+    pub source_exchange_routing_key: Option<&'a str>,
+}
+
+impl<'a> Amqp091ShovelSourceParams<'a> {
+    pub fn queue_source(source_uri: &'a str, source_queue: &'a str) -> Self {
+        Self {
+            source_uri,
+            source_queue: Some(source_queue),
+
+            source_exchange: None,
+            source_exchange_routing_key: None,
+        }
+    }
+
+    pub fn exchange_source(
+        source_uri: &'a str,
+        source_exchange: &'a str,
+        source_exchange_routing_key: Option<&'a str>,
+    ) -> Self {
+        Self {
+            source_uri,
+            source_exchange: Some(source_exchange),
+            source_exchange_routing_key,
+
+            source_queue: None,
+        }
+    }
+}
+
+#[derive(Serialize)]
+pub struct Amqp091ShovelDestinationParams<'a> {
+    pub destination_uri: &'a str,
+
+    pub destination_queue: Option<&'a str>,
+    pub destination_exchange: Option<&'a str>,
+    pub destination_exchange_routing_key: Option<&'a str>,
+}
+
+impl<'a> Amqp091ShovelDestinationParams<'a> {
+    pub fn queue_destination(destination_uri: &'a str, destination_queue: &'a str) -> Self {
+        Self {
+            destination_uri,
+            destination_queue: Some(destination_queue),
+
+            destination_exchange: None,
+            destination_exchange_routing_key: None,
+        }
+    }
+
+    pub fn exchange_destination(
+        destination_uri: &'a str,
+        destination_exchange: &'a str,
+        destination_exchange_routing_key: Option<&'a str>,
+    ) -> Self {
+        Self {
+            destination_uri,
+            destination_exchange: Some(destination_exchange),
+            destination_exchange_routing_key,
+
+            destination_queue: None,
+        }
+    }
+}
+
 pub type MessageProperties = Map<String, Value>;
 
 #[derive(Serialize, Default)]

tests/async_consumer_tests.rs

@@ -36,4 +36,6 @@ async fn test_async_list_vhost_consumers() {
 
     let result2 = rc.list_consumers_in(vh_params.name).await;
     assert!(result2.is_ok(), "list_consumers_in returned {:?}", result2);
+
+    rc.delete_vhost(&vh_params.name, true).await.unwrap();
 }