Introduce Client#import_vhost_definitions

to import virtual host definitions
(as opposed to cluster-wide ones)

Author: michaelklishin

CHANGELOG.md

@@ -1,11 +1,22 @@
 # Rust Client for the RabbitMQ HTTP API Change Log
 
-## v0.22.0  (in development)
+## v0.23.0  (in development)
 
 No (documented) changes yet.
 
 
-## v0.21.0  (Feb 8, 2025)
+## v0.22.0 (Feb 8, 2025)
+
+### Enhancements
+
+ * `Client#import_vhost_definitions` is a new function that imports virtual host-specific
+   definition files (as opposed to cluster-wide ones) into the target virtual host
+
+ * `Client#import_cluster_wide_definitions` is an alias to `Client#import_definitions`
+   to better reflect what it does
+
+
+## v0.21.0 (Feb 8, 2025)
 
 ### Enhancements
 

README.md

@@ -14,25 +14,25 @@ This library is relatively young, breaking API changes are possible.
 ### Blocking Client
 
 ```toml
-rabbitmq_http_client = { version = "0.21.0", features = ["core", "blocking"] }
+rabbitmq_http_client = { version = "0.22.0", features = ["core", "blocking"] }
 ```
 
 ### Async Client
 
 ```toml
-rabbitmq_http_client = { version = "0.21.0", features = ["core", "async"] }
+rabbitmq_http_client = { version = "0.22.0", features = ["core", "async"] }
 ```
 
 ### Blocking Client with Tabled Support
 
 ```toml
-rabbitmq_http_client = { version = "0.21.0", features = ["core", "blocking", "tabled"] }
+rabbitmq_http_client = { version = "0.22.0", features = ["core", "blocking", "tabled"] }
 ```
 
 ### Async Client with Tabled Support
 
 ```toml
-rabbitmq_http_client = { version = "0.21.0", features = ["core", "async", "tabled"] }
+rabbitmq_http_client = { version = "0.22.0", features = ["core", "async", "tabled"] }
 ```
 
 

src/api.rs

@@ -1296,11 +1296,21 @@ where
     }
 
     pub async fn import_definitions(&self, definitions: Value) -> Result<()> {
+        self.import_cluster_wide_definitions(definitions).await
+    }
+
+    pub async fn import_cluster_wide_definitions(&self, definitions: Value) -> Result<()> {
         self.http_post("definitions", &definitions, None, None)
             .await?;
         Ok(())
     }
 
+    pub async fn import_vhost_definitions(&self, vhost: &str, definitions: Value) -> Result<()> {
+        self.http_post(path!("definitions", vhost), &definitions, None, None)
+            .await?;
+        Ok(())
+    }
+
     //
     // Health Checks
     //

src/blocking_api.rs

@@ -1133,10 +1133,19 @@ where
     }
 
     pub fn import_definitions(&self, definitions: Value) -> Result<()> {
+        self.import_cluster_wide_definitions(definitions)
+    }
+
+    pub fn import_cluster_wide_definitions(&self, definitions: Value) -> Result<()> {
         self.http_post("definitions", &definitions, None, None)?;
         Ok(())
     }
 
+    pub fn import_vhost_definitions(&self, vhost: &str, definitions: Value) -> Result<()> {
+        self.http_post(path!("definitions", vhost), &definitions, None, None)?;
+        Ok(())
+    }
+
     //
     // Health Checks
     //

tests/async_definitions_tests.rs

@@ -14,7 +14,9 @@
 use rabbitmq_http_client::api::Client;
 
 mod test_helpers;
-use crate::test_helpers::{await_metric_emission, endpoint, PASSWORD, USERNAME};
+use crate::test_helpers::{
+    await_metric_emission, await_queue_metric_emission, endpoint, PASSWORD, USERNAME,
+};
 use rabbitmq_http_client::commons::PolicyTarget;
 use rabbitmq_http_client::requests::{
     ExchangeParams, PolicyParams, QueueParams, VirtualHostParams,
@@ -227,7 +229,7 @@ async fn test_async_export_vhost_definitions_as_data() {
 }
 
 #[tokio::test]
-async fn test_async_import_definitions() {
+async fn test_async_import_cluster_definitions() {
     let endpoint = endpoint();
     let rc = Client::new(&endpoint, USERNAME, PASSWORD);
     let _ = rc.delete_queue("/", "imported_queue", false).await;
@@ -240,8 +242,12 @@ async fn test_async_import_definitions() {
       }
     ]});
 
-    let result = rc.import_definitions(defs).await;
-    assert!(result.is_ok(), "import_definitions returned {:?}", result);
+    let result = rc.import_cluster_wide_definitions(defs).await;
+    assert!(
+        result.is_ok(),
+        "import_cluster_wide_definitions returned {:?}",
+        result
+    );
 
     let result1 = rc.get_queue_info("/", "imported_queue").await;
     assert!(
@@ -250,3 +256,42 @@ async fn test_async_import_definitions() {
         result1
     );
 }
+
+#[tokio::test]
+async fn test_async_import_vhost_definitions() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint, USERNAME, PASSWORD);
+
+    let vh = "rust/http/api/async/vhost.definitions.import";
+    let _ = rc.delete_vhost(vh, true).await.unwrap();
+
+    let vh_params = VirtualHostParams::named(vh);
+    rc.create_vhost(&vh_params).await.unwrap();
+
+    let q = "imported_queue";
+    let defs = json!({  "queues": [
+      {
+        "auto_delete": false,
+        "durable": true,
+        "name": q,
+      }
+    ]});
+
+    let result = rc.import_vhost_definitions(vh, defs).await;
+    assert!(
+        result.is_ok(),
+        "import_vhost_definitions returned {:?}",
+        result
+    );
+
+    await_queue_metric_emission();
+
+    let result1 = rc.get_queue_info(vh, q).await;
+    assert!(
+        result1.is_ok(),
+        "can't get the imported queue: {:?}",
+        result1
+    );
+
+    let _ = rc.delete_vhost(vh, true).await.unwrap();
+}

tests/blocking_definitions_tests.rs

@@ -14,7 +14,9 @@
 use rabbitmq_http_client::blocking_api::Client;
 
 mod test_helpers;
-use crate::test_helpers::{await_metric_emission, endpoint, PASSWORD, USERNAME};
+use crate::test_helpers::{
+    await_metric_emission, await_queue_metric_emission, endpoint, PASSWORD, USERNAME,
+};
 use rabbitmq_http_client::commons::PolicyTarget;
 use rabbitmq_http_client::requests::{
     ExchangeParams, PolicyParams, QueueParams, VirtualHostParams,
@@ -214,26 +216,73 @@ fn test_blocking_export_vhost_definitions_as_data() {
 }
 
 #[test]
-fn test_blocking_import_definitions() {
+fn test_blocking_import_cluster_definitions() {
     let endpoint = endpoint();
     let rc = Client::new(&endpoint, USERNAME, PASSWORD);
-    let _ = rc.delete_queue("/", "imported_queue", false);
+    let q = "imported_queue";
+
+    let _ = rc.delete_queue("/", q, false);
     let defs = json!({  "queues": [
       {
         "auto_delete": false,
         "durable": true,
-        "name": "imported_queue",
+        "name": q,
         "vhost": "/"
       }
     ]});
 
-    let result = rc.import_definitions(defs);
-    assert!(result.is_ok(), "import_definitions returned {:?}", result);
+    let result = rc.import_cluster_wide_definitions(defs);
+    assert!(
+        result.is_ok(),
+        "import_cluster_wide_definitions returned {:?}",
+        result
+    );
 
-    let result1 = rc.get_queue_info("/", "imported_queue");
+    let result1 = rc.get_queue_info("/", q);
+    assert!(
+        result1.is_ok(),
+        "can't get the imported import_cluster_wide_definitions: {:?}",
+        result1
+    );
+
+    rc.delete_queue("/", q, true).unwrap();
+}
+
+#[test]
+fn test_blocking_import_vhost_definitions() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint, USERNAME, PASSWORD);
+
+    let vh = "rust/http/api/blocking/vhost.definitions.import";
+    let _ = rc.delete_vhost(vh, true).unwrap();
+
+    let vh_params = VirtualHostParams::named(vh);
+    rc.create_vhost(&vh_params).unwrap();
+
+    let q = "imported_queue";
+    let defs = json!({  "queues": [
+      {
+        "auto_delete": false,
+        "durable": true,
+        "name": q,
+      }
+    ]});
+
+    let result = rc.import_vhost_definitions(vh, defs);
+    assert!(
+        result.is_ok(),
+        "import_vhost_definitions returned {:?}",
+        result
+    );
+
+    await_queue_metric_emission();
+
+    let result1 = rc.get_queue_info(vh, q);
     assert!(
         result1.is_ok(),
         "can't get the imported queue: {:?}",
         result1
     );
+
+    let _ = rc.delete_vhost(vh, true).unwrap();
 }