Support for GET /api/queues/detailed

Author: michaelklishin

src/api.rs

@@ -20,9 +20,10 @@ use crate::requests::{
     FederationUpstreamParams, GlobalRuntimeParameterDefinition, SHOVEL_COMPONENT, StreamParams,
 };
 use crate::responses::{
-    AuthenticationAttemptStatistics, ClusterTags, DeprecatedFeatureList, FeatureFlag, FeatureFlagList, FeatureFlagStability,
-    FeatureFlagState, FederationUpstream, GetMessage, OAuthConfiguration, Overview,
-    SchemaDefinitionSyncStatus, VirtualHostDefinitionSet, WarmStandbyReplicationStatus,
+    AuthenticationAttemptStatistics, ClusterTags, DeprecatedFeatureList, FeatureFlag,
+    FeatureFlagList, FeatureFlagStability, FeatureFlagState, FederationUpstream, GetMessage,
+    OAuthConfiguration, Overview, SchemaDefinitionSyncStatus, VirtualHostDefinitionSet,
+    WarmStandbyReplicationStatus,
 };
 use crate::{
     commons::{BindingDestinationType, SupportedProtocol, UserLimitTarget, VirtualHostLimitTarget},
@@ -520,6 +521,14 @@ where
         Ok(response)
     }
 
+    /// Lists all queues and streams across the cluster. Compared to [`list_queues`], provides more queue metrics.
+    /// See [Queues Guide](https://www.rabbitmq.com/docs/queues) and [RabbitMQ Streams Guide](https://www.rabbitmq.com/docs/streams) to learn more.
+    pub async fn list_queues_with_details(&self) -> Result<Vec<responses::DetailedQueueInfo>> {
+        let response = self.http_get("queues/detailed", None, None).await?;
+        let response = response.json().await?;
+        Ok(response)
+    }
+
     /// Lists all exchanges across the cluster.
     /// See [Exchanges Guide](https://www.rabbitmq.com/docs/exchanges) to learn more.
     pub async fn list_exchanges(&self) -> Result<Vec<responses::ExchangeInfo>> {
@@ -1883,8 +1892,13 @@ where
     }
 
     /// Returns authentication attempt statistics for a given node.
-    pub async fn auth_attempts_statistics(&self, node: &str) -> Result<Vec<AuthenticationAttemptStatistics>> {
-        let response = self.http_get(path!("auth", "attempts", node), None, None).await?;
+    pub async fn auth_attempts_statistics(
+        &self,
+        node: &str,
+    ) -> Result<Vec<AuthenticationAttemptStatistics>> {
+        let response = self
+            .http_get(path!("auth", "attempts", node), None, None)
+            .await?;
         let response = response.json().await?;
         Ok(response)
     }

src/blocking_api.rs

@@ -20,9 +20,9 @@ use crate::requests::{
     FederationUpstreamParams, GlobalRuntimeParameterDefinition, SHOVEL_COMPONENT, StreamParams,
 };
 use crate::responses::{
-    AuthenticationAttemptStatistics, ClusterTags, DeprecatedFeatureList, FeatureFlag, FeatureFlagList, FeatureFlagStability,
-    FeatureFlagState, FederationUpstream, GetMessage, OAuthConfiguration, VirtualHostDefinitionSet,
-    WarmStandbyReplicationStatus,
+    AuthenticationAttemptStatistics, ClusterTags, DeprecatedFeatureList, FeatureFlag,
+    FeatureFlagList, FeatureFlagStability, FeatureFlagState, FederationUpstream, GetMessage,
+    OAuthConfiguration, VirtualHostDefinitionSet, WarmStandbyReplicationStatus,
 };
 use crate::{
     commons::{BindingDestinationType, SupportedProtocol, UserLimitTarget, VirtualHostLimitTarget},
@@ -466,6 +466,12 @@ where
         self.get_api_request(path!("queues", virtual_host))
     }
 
+    /// Lists all queues and streams across the cluster. Compared to [`list_queues`], provides more queue metrics.
+    /// See [Queues Guide](https://www.rabbitmq.com/docs/queues) and [RabbitMQ Streams Guide](https://www.rabbitmq.com/docs/streams) to learn more.
+    pub fn list_queues_with_details(&self) -> Result<Vec<responses::DetailedQueueInfo>> {
+        self.get_api_request("queues/detailed")
+    }
+
     /// Lists all exchanges across the cluster.
     /// See [Exchanges Guide](https://www.rabbitmq.com/docs/exchanges) to learn more.
     pub fn list_exchanges(&self) -> Result<Vec<responses::ExchangeInfo>> {
@@ -1749,7 +1755,10 @@ where
     }
 
     /// Returns authentication attempt statistics for a given node.
-    pub fn auth_attempts_statistics(&self, node: &str) -> Result<Vec<AuthenticationAttemptStatistics>> {
+    pub fn auth_attempts_statistics(
+        &self,
+        node: &str,
+    ) -> Result<Vec<AuthenticationAttemptStatistics>> {
         let response = self.http_get(path!("auth", "attempts", node), None, None)?;
         let response = response.json()?;
         Ok(response)

src/responses.rs

@@ -19,8 +19,8 @@ use std::{fmt, ops};
 
 use crate::commons::{
     BindingDestinationType, ChannelId, MessageTransferAcknowledgementMode, OverflowBehavior,
-    PolicyTarget, QueueType, SupportedProtocol, Username, VirtualHostName, X_ARGUMENT_KEY_X_OVERFLOW,
-    X_ARGUMENT_KEY_X_QUEUE_TYPE,
+    PolicyTarget, QueueType, SupportedProtocol, Username, VirtualHostName,
+    X_ARGUMENT_KEY_X_OVERFLOW, X_ARGUMENT_KEY_X_QUEUE_TYPE,
 };
 use crate::error::ConversionError;
 use crate::formatting::*;
@@ -979,6 +979,184 @@ pub struct QueueInfo {
     pub unacknowledged_message_count: u64,
 }
 
+/// Represents detailed queue information with extended metrics and garbage collection details.
+/// This is an enhanced version of `QueueInfo` that includes additional fields from the detailed queues endpoint.
+#[derive(Debug, Deserialize, Clone)]
+#[cfg_attr(feature = "tabled", derive(Tabled))]
+#[allow(dead_code)]
+pub struct DetailedQueueInfo {
+    pub name: String,
+    pub vhost: VirtualHostName,
+    #[serde(rename(deserialize = "type"))]
+    pub queue_type: String,
+    pub durable: bool,
+    pub auto_delete: bool,
+    pub exclusive: bool,
+    #[cfg_attr(feature = "tabled", tabled(display = "display_arg_table"))]
+    pub arguments: XArguments,
+
+    #[serde(default = "undefined")]
+    pub node: String,
+    #[serde(default)]
+    pub state: String,
+    // only quorum queues and streams will have this
+    #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
+    pub leader: Option<String>,
+    #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
+    pub members: Option<NodeList>,
+    #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
+    pub online: Option<NodeList>,
+
+    #[serde(default)]
+    pub memory: u64,
+    #[serde(rename(deserialize = "consumers"))]
+    #[serde(default)]
+    pub consumer_count: u16,
+    #[serde(default)]
+    pub consumer_utilisation: f32,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub exclusive_consumer_tag: Option<String>,
+
+    #[cfg_attr(feature = "tabled", tabled(display = "display_option"))]
+    pub policy: Option<String>,
+
+    #[serde(default)]
+    pub message_bytes: u64,
+    #[serde(default)]
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub message_bytes_persistent: u64,
+    #[serde(default)]
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub message_bytes_ram: u64,
+    #[serde(default)]
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub message_bytes_ready: u64,
+    #[serde(default)]
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub message_bytes_unacknowledged: u64,
+
+    #[serde(rename(deserialize = "messages"))]
+    #[serde(default)]
+    pub message_count: u64,
+    #[serde(rename(deserialize = "messages_persistent"))]
+    #[serde(default)]
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub on_disk_message_count: u64,
+    #[serde(rename(deserialize = "messages_ram"))]
+    #[serde(default)]
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub in_memory_message_count: u64,
+    #[serde(rename(deserialize = "messages_unacknowledged"))]
+    #[serde(default)]
+    pub unacknowledged_message_count: u64,
+
+    // Additional detailed fields
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub garbage_collection: Option<GarbageCollectionDetails>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_batch_size: Option<u32>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_batch_size_avg: Option<f64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_batch_size_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_file_handle_open_attempt_avg_time: Option<f64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_file_handle_open_attempt_avg_time_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_read_avg_time: Option<f64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_read_avg_time_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_read_bytes: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_read_bytes_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_read_count: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_read_count_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_reopen_count: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_reopen_count_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_seek_avg_time: Option<f64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_seek_avg_time_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_seek_count: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_seek_count_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_sync_avg_time: Option<f64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_sync_avg_time_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_sync_count: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_sync_count_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_write_avg_time: Option<f64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_write_avg_time_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_write_bytes: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_write_bytes_details: Option<Rate>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_write_count: Option<u64>,
+    #[cfg_attr(feature = "tabled", tabled(skip))]
+    pub io_write_count_details: Option<Rate>,
+}
+
+/// Garbage collection details for queue processes
+#[derive(Debug, Deserialize, Clone)]
+#[cfg_attr(feature = "tabled", derive(Tabled))]
+#[allow(dead_code)]
+pub struct GarbageCollectionDetails {
+    pub fullsweep_after: u32,
+    pub max_heap_size: u32,
+    pub min_bin_vheap_size: u32,
+    pub min_heap_size: u32,
+    pub minor_gcs: u32,
+}
+
+impl QueueOps for DetailedQueueInfo {
+    fn name(&self) -> &str {
+        &self.name
+    }
+
+    fn queue_type(&self) -> QueueType {
+        QueueType::from(self.queue_type.as_str())
+    }
+
+    fn policy_target_type(&self) -> PolicyTarget {
+        PolicyTarget::from(self.queue_type())
+    }
+
+    fn x_arguments(&self) -> &XArguments {
+        &self.arguments
+    }
+}
+
+impl NamedPolicyTargetObject for DetailedQueueInfo {
+    fn vhost(&self) -> String {
+        self.vhost.clone()
+    }
+
+    fn name(&self) -> String {
+        self.name.clone()
+    }
+
+    fn policy_target(&self) -> PolicyTarget {
+        self.policy_target_type()
+    }
+
+    fn does_match(&self, policy: &Policy) -> bool {
+        policy.does_match_object(self)
+    }
+}
+
 impl QueueOps for QueueInfo {
     fn name(&self) -> &str {
         &self.name

tests/async_auth_attempts_tests.rs

@@ -39,7 +39,7 @@ pub async fn test_async_auth_attempts_statistics() {
                 for proto in stats {
                     assert!(proto.failure_count + proto.success_count <= proto.all_attempt_count);
                 }
-            },
+            }
             Err(_) => {
                 // Assume the endpoint wasn't available in this RabbitMQ version
             }

tests/async_queue_tests.rs

@@ -146,3 +146,49 @@ async fn test_async_list_queues_in_a_virtual_host() {
 
     rc.delete_queue(vh_name, params.name, false).await.unwrap();
 }
+
+#[tokio::test]
+async fn test_async_list_queues_with_details() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint, USERNAME, PASSWORD);
+
+    let vh_name = "/";
+
+    let params = QueueParams::new_durable_classic_queue("rust.tests.cq.detailed.92734827364", None);
+    let result1 = rc.declare_queue(vh_name, &params).await;
+    assert!(result1.is_ok(), "declare_queue returned {result1:?}");
+
+    test_helpers::async_await_queue_metric_emission().await;
+
+    let result2 = rc.list_queues_with_details().await;
+    assert!(
+        result2.is_ok(),
+        "list_queues_with_details returned {result2:?}"
+    );
+
+    let detailed_queues = result2.unwrap();
+    assert!(
+        !detailed_queues.is_empty(),
+        "Expected at least one queue in detailed list"
+    );
+
+    // Find our test queue in the results
+    let test_queue = detailed_queues.iter().find(|q| q.name == params.name);
+    assert!(
+        test_queue.is_some(),
+        "Expected to find our test queue in detailed results"
+    );
+
+    let queue = test_queue.unwrap();
+    // Verify basic queue properties are present
+    assert_eq!(queue.name, params.name);
+    assert_eq!(queue.vhost, vh_name);
+    assert_eq!(queue.durable, true);
+
+    // More fields
+    if let Some(gc) = &queue.garbage_collection {
+        assert!(gc.fullsweep_after > 1000);
+    }
+
+    rc.delete_queue(vh_name, params.name, false).await.unwrap();
+}

tests/blocking_auth_attempts_tests.rs

@@ -39,7 +39,7 @@ pub fn test_blocking_auth_attempts_statistics() {
                 for proto in stats {
                     assert!(proto.failure_count + proto.success_count <= proto.all_attempt_count);
                 }
-            },
+            }
             Err(_) => {
                 // Assume the endpoint wasn't available in this RabbitMQ version
             }

tests/blocking_queue_tests.rs

@@ -146,3 +146,50 @@ fn test_blocking_list_queues_in_a_virtual_host() {
 
     rc.delete_queue(vh_name, params.name, false).unwrap();
 }
+
+#[test]
+pub fn test_blocking_list_queues_with_details() {
+    let endpoint = endpoint();
+    let rc = Client::new(&endpoint, USERNAME, PASSWORD);
+
+    let vh_name = "/";
+
+    let params =
+        QueueParams::new_durable_classic_queue("rust.tests.cq.detailed.blocking.18273486", None);
+    let result1 = rc.declare_queue(vh_name, &params);
+    assert!(result1.is_ok(), "declare_queue returned {result1:?}");
+
+    test_helpers::await_queue_metric_emission();
+
+    let result2 = rc.list_queues_with_details();
+    assert!(
+        result2.is_ok(),
+        "list_queues_with_details returned {result2:?}"
+    );
+
+    let detailed_queues = result2.unwrap();
+    assert!(
+        !detailed_queues.is_empty(),
+        "Expected at least one queue in detailed list"
+    );
+
+    // Find our test queue in the results
+    let test_queue = detailed_queues.iter().find(|q| q.name == params.name);
+    assert!(
+        test_queue.is_some(),
+        "Expected to find our test queue in detailed results"
+    );
+
+    let queue = test_queue.unwrap();
+    // Verify basic queue properties are present
+    assert_eq!(queue.name, params.name);
+    assert_eq!(queue.vhost, vh_name);
+    assert_eq!(queue.durable, true);
+
+    // More fields
+    if let Some(gc) = &queue.garbage_collection {
+        assert!(gc.fullsweep_after > 1000);
+    }
+
+    rc.delete_queue(vh_name, params.name, false).unwrap();
+}