Merge pull request #8 from michaelklishin/mk-limits

Implement ops on virtual host limits

src/blocking.rs

@@ -1,7 +1,8 @@
 use crate::{
+    commons::EnforcedLimitTarget,
     requests::{
         ExchangeParams, PolicyParams, QueueParams, RuntimeParameterDefinition, UserParams,
-        VirtualHostParams, XArguments,
+        VirtualHostLimitParams, VirtualHostParams, XArguments,
     },
     responses,
 };
@@ -479,6 +480,50 @@ impl<'a> Client<'a> {
         Ok(())
     }
 
+    pub fn clear_all_runtime_parameters_of_component(&self, component: &str) -> Result<()> {
+        let params = self.list_runtime_parameters_of_component(component)?;
+        for rp in params {
+            self.clear_runtime_parameter(&rp.component, &rp.vhost, &rp.name)?
+        }
+        Ok(())
+    }
+
+    pub fn set_vhost_limit(&self, vhost: &str, limit: VirtualHostLimitParams) -> Result<()> {
+        let path = format!("vhost-limits/{}/{}", vhost, String::from(limit.kind));
+
+        let mut body = Map::<String, Value>::new();
+        body.insert("value".to_owned(), json!(limit.value));
+
+        let response = self.http_put(&path, &body)?;
+        self.ok_or_status_code_error(response)?;
+        Ok(())
+    }
+
+    pub fn clear_vhost_limit(&self, vhost: &str, kind: EnforcedLimitTarget) -> Result<()> {
+        let path = format!("vhost-limits/{}/{}", vhost, String::from(kind));
+
+        let response = self.http_delete(&path)?;
+        self.ok_or_status_code_error(response)?;
+        Ok(())
+    }
+
+    pub fn list_all_vhost_limits(&self) -> Result<Vec<responses::VirtualHostLimits>> {
+        let response = self.http_get("vhost-limits")?;
+        let response2 = self.ok_or_status_code_error(response)?;
+        response2
+            .json::<Vec<responses::VirtualHostLimits>>()
+            .map_err(Error::from)
+    }
+
+    pub fn list_vhost_limits(&self, vhost: &str) -> Result<Vec<responses::VirtualHostLimits>> {
+        let path = format!("vhost-limits/{}", vhost);
+        let response = self.http_get(&path)?;
+        let response2 = self.ok_or_status_code_error(response)?;
+        response2
+            .json::<Vec<responses::VirtualHostLimits>>()
+            .map_err(Error::from)
+    }
+
     pub fn get_cluster_name(&self) -> Result<responses::ClusterIdentity> {
         let response = self.http_get("cluster-name")?;
         let response2 = self.ok_or_status_code_error(response)?;

src/commons.rs

@@ -204,3 +204,48 @@ impl From<PolicyTarget> for String {
         }
     }
 }
+
+#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+pub enum EnforcedLimitTarget {
+    MaxConnections,
+    MaxQueues,
+}
+
+impl ToString for EnforcedLimitTarget {
+    fn to_string(&self) -> String {
+        match self {
+            EnforcedLimitTarget::MaxConnections => "max-connections".to_owned(),
+            EnforcedLimitTarget::MaxQueues => "max-queues".to_owned(),
+        }
+    }
+}
+
+impl From<&str> for EnforcedLimitTarget {
+    fn from(value: &str) -> Self {
+        match value {
+            "max-connections" => EnforcedLimitTarget::MaxConnections,
+            "max-queues" => EnforcedLimitTarget::MaxQueues,
+            _ => EnforcedLimitTarget::MaxConnections,
+        }
+    }
+}
+
+impl From<String> for EnforcedLimitTarget {
+    fn from(value: String) -> Self {
+        match value.as_str() {
+            "max-connections" => EnforcedLimitTarget::MaxConnections,
+            "max-queues" => EnforcedLimitTarget::MaxQueues,
+            _ => EnforcedLimitTarget::MaxConnections,
+        }
+    }
+}
+
+impl From<EnforcedLimitTarget> for String {
+    fn from(value: EnforcedLimitTarget) -> Self {
+        match value {
+            EnforcedLimitTarget::MaxConnections => "max-connections".to_owned(),
+            EnforcedLimitTarget::MaxQueues => "max-queues".to_owned(),
+        }
+    }
+}

src/requests.rs

@@ -1,4 +1,4 @@
-use crate::commons::{ExchangeType, PolicyTarget, QueueType};
+use crate::commons::{EnforcedLimitTarget, ExchangeType, PolicyTarget, QueueType};
 use serde::Serialize;
 use serde_json::{Map, Value};
 
@@ -14,6 +14,30 @@ pub struct VirtualHostParams<'a> {
     pub tracing: bool,
 }
 
+impl<'a> VirtualHostParams<'a> {
+    pub fn named(name: &'a str) -> Self {
+        VirtualHostParams {
+            name,
+            description: None,
+            tags: None,
+            default_queue_type: None,
+            tracing: false,
+        }
+    }
+}
+
+#[derive(Serialize)]
+pub struct VirtualHostLimitParams {
+    pub kind: EnforcedLimitTarget,
+    pub value: i64,
+}
+
+impl VirtualHostLimitParams {
+    pub fn new(kind: EnforcedLimitTarget, value: i64) -> Self {
+        VirtualHostLimitParams { kind, value }
+    }
+}
+
 #[derive(Serialize)]
 pub struct UserParams<'a> {
     pub name: &'a str,

src/responses.rs

@@ -21,6 +21,16 @@ pub struct VirtualHost {
     pub metadata: VirtualHostMetadata,
 }
 
+pub type EnforcedLimits = Map<String, serde_json::Value>;
+
+#[derive(Debug, Deserialize, Clone)]
+#[allow(dead_code)]
+pub struct VirtualHostLimits {
+    pub vhost: String,
+    #[serde(rename(deserialize = "value"))]
+    pub limits: EnforcedLimits,
+}
+
 #[derive(Debug, Deserialize, Clone)]
 #[allow(dead_code)]
 pub struct User {

tests/runtime_parameters_tests.rs

@@ -69,7 +69,10 @@ fn test_clear_runtime_parameter() {
     let result4 = rc.list_runtime_parameters();
     assert!(result4.is_ok());
     let vec = result4.unwrap();
-    assert!(vec.is_empty());
+    assert!(vec
+        .iter()
+        .find(|p| p.component == "vhost-limits" && p.vhost == vh_params.name.to_owned())
+        .is_none());
 
     let _ = rc.delete_vhost(vh_params.name);
 }

tests/virtual_host_limit_tests.rs

@@ -0,0 +1,71 @@
+use rabbitmq_http_client::{
+    blocking::Client,
+    commons::EnforcedLimitTarget,
+    requests::{VirtualHostLimitParams, VirtualHostParams},
+};
+
+mod common;
+use crate::common::{endpoint, PASSWORD, USERNAME};
+
+#[test]
+fn test_list_all_vhost_limits() {
+    let endpoint = endpoint();
+    let rc = Client::new_with_basic_auth_credentials(&endpoint, USERNAME, PASSWORD);
+
+    let vh_params = VirtualHostParams::named("test_list_all_vhost_limits");
+    let result1 = rc.create_vhost(&vh_params);
+    assert!(result1.is_ok());
+
+    let limit = VirtualHostLimitParams::new(EnforcedLimitTarget::MaxQueues, 500);
+    let result2 = rc.set_vhost_limit(&vh_params.name, limit);
+    assert!(result2.is_ok());
+
+    let result3 = rc.list_all_vhost_limits();
+    assert!(result3.is_ok());
+    let vec = result3.unwrap();
+    assert!(vec.iter().find(|vh| vh.vhost == vh_params.name).is_some());
+
+    let key1 = EnforcedLimitTarget::MaxConnections.to_string();
+    assert!(vec
+        .iter()
+        .find(|it| it.vhost == vh_params.name && it.limits.get(&key1).is_some())
+        .is_none());
+    let key2 = EnforcedLimitTarget::MaxQueues.to_string();
+    assert!(vec
+        .iter()
+        .find(|it| it.vhost == vh_params.name && it.limits.get(&key2).is_some())
+        .is_some());
+
+    rc.delete_vhost(vh_params.name).unwrap();
+}
+
+#[test]
+fn test_list_vhost_limits() {
+    let endpoint = endpoint();
+    let rc = Client::new_with_basic_auth_credentials(&endpoint, USERNAME, PASSWORD);
+
+    let vh_params = VirtualHostParams::named("test_list_vhost_limits");
+    let result1 = rc.create_vhost(&vh_params);
+    assert!(result1.is_ok());
+
+    let limit = VirtualHostLimitParams::new(EnforcedLimitTarget::MaxConnections, 500);
+    let result2 = rc.set_vhost_limit(&vh_params.name, limit);
+    assert!(result2.is_ok());
+
+    let result3 = rc.list_vhost_limits(vh_params.name);
+    assert!(result3.is_ok());
+    let vec = result3.unwrap();
+
+    let key1 = EnforcedLimitTarget::MaxConnections.to_string();
+    assert!(vec
+        .iter()
+        .find(|it| it.vhost == vh_params.name && it.limits.get(&key1).is_some())
+        .is_some());
+    let key2 = EnforcedLimitTarget::MaxQueues.to_string();
+    assert!(vec
+        .iter()
+        .find(|it| it.vhost == vh_params.name && it.limits.get(&key2).is_some())
+        .is_none());
+
+    rc.delete_vhost(vh_params.name).unwrap();
+}