feat(rabbitmq): show sortable queue message counts

Closes #4286

crates/dbx-core/src/mq/adapters/kafka.rs

@@ -203,6 +203,9 @@ impl MessageQueueAdmin for KafkaAdmin {
                     internal: t.get("internal").and_then(|v| v.as_bool()).unwrap_or(false),
                     message_type: None,
                     namespace: None,
+                    message_count: None,
+                    messages_ready: None,
+                    messages_unacked: None,
                 }
             })
             .collect())

crates/dbx-core/src/mq/adapters/pulsar.rs

@@ -297,6 +297,9 @@ impl MessageQueueAdmin for PulsarAdmin {
                         internal: false,
                         message_type: None,
                         namespace: None,
+                        message_count: None,
+                        messages_ready: None,
+                        messages_unacked: None,
                     })
                 })
                 .buffered(PARTITION_METADATA_CONCURRENCY)
@@ -323,6 +326,9 @@ impl MessageQueueAdmin for PulsarAdmin {
                     internal: false,
                     message_type: None,
                     namespace: None,
+                    message_count: None,
+                    messages_ready: None,
+                    messages_unacked: None,
                 });
             }
         }

crates/dbx-core/src/mq/adapters/rabbitmq.rs

@@ -183,23 +183,7 @@ impl MessageQueueAdmin for RabbitMqAdmin {
         let result: serde_json::Value = self.call("mq_list_topics", params).await?;
         let topics = result.get("topics").and_then(|v| v.as_array()).cloned().unwrap_or_default();
 
-        Ok(topics
-            .into_iter()
-            .map(|t| {
-                let name = t.get("name").and_then(|v| v.as_str()).unwrap_or_default().to_string();
-                TopicInfo {
-                    name: name.clone(),
-                    short_name: name,
-                    partitioned: false,
-                    partitions: None,
-                    persistent: t.get("durable").and_then(|v| v.as_bool()).unwrap_or(true),
-                    internal: t.get("internal").and_then(|v| v.as_bool()).unwrap_or(false),
-                    message_type: None,
-                    // All-vhosts listings report each queue's own vhost.
-                    namespace: t.get("vhost").and_then(|v| v.as_str()).map(String::from),
-                }
-            })
-            .collect())
+        Ok(topics.into_iter().map(|topic| topic_info_from_agent_value(&topic)).collect())
     }
 
     async fn create_topic(&self, topic: &TopicRef, _partitions: Option<u32>) -> Result<(), String> {
@@ -1013,6 +997,24 @@ fn peeked_message_from_json(idx: usize, m: &serde_json::Value) -> PeekedMessage
     }
 }
 
+fn topic_info_from_agent_value(topic: &serde_json::Value) -> TopicInfo {
+    let name = topic.get("name").and_then(|value| value.as_str()).unwrap_or_default().to_string();
+    TopicInfo {
+        name: name.clone(),
+        short_name: name,
+        partitioned: false,
+        partitions: None,
+        persistent: topic.get("durable").and_then(|value| value.as_bool()).unwrap_or(true),
+        internal: topic.get("internal").and_then(|value| value.as_bool()).unwrap_or(false),
+        message_type: None,
+        // All-vhosts listings report each queue's own vhost.
+        namespace: topic.get("vhost").and_then(|value| value.as_str()).map(String::from),
+        message_count: topic.get("messages").and_then(|value| value.as_i64()),
+        messages_ready: topic.get("messagesReady").and_then(|value| value.as_i64()),
+        messages_unacked: topic.get("messagesUnacked").and_then(|value| value.as_i64()),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1188,6 +1190,23 @@ mod tests {
         assert_eq!(queue_name(&topic), "orders.queue");
     }
 
+    #[test]
+    fn topic_info_maps_queue_message_counts() {
+        let topic = topic_info_from_agent_value(&serde_json::json!({
+            "name": "orders",
+            "durable": true,
+            "vhost": "/",
+            "messages": 12,
+            "messagesReady": 10,
+            "messagesUnacked": 2
+        }));
+
+        assert_eq!(topic.message_count, Some(12));
+        assert_eq!(topic.messages_ready, Some(10));
+        assert_eq!(topic.messages_unacked, Some(2));
+        assert_eq!(topic.namespace.as_deref(), Some("/"));
+    }
+
     #[test]
     fn peeked_message_maps_agent_fields() {
         let msg = serde_json::json!({

crates/dbx-core/src/mq/adapters/rocketmq.rs

@@ -764,6 +764,9 @@ fn topic_info_from_agent_value(t: &serde_json::Value) -> TopicInfo {
         internal: t.get("internal").and_then(|v| v.as_bool()).unwrap_or(false),
         message_type: t.get("messageType").and_then(|v| v.as_str()).map(String::from),
         namespace: None,
+        message_count: None,
+        messages_ready: None,
+        messages_unacked: None,
     }
 }
 

crates/dbx-core/src/mq/types.rs

@@ -331,6 +331,15 @@ pub struct TopicInfo {
     /// cross-namespace listings such as the RabbitMQ "all vhosts" mode.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub namespace: Option<String>,
+    /// RabbitMQ total queue messages, including ready and unacknowledged messages.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub message_count: Option<i64>,
+    /// RabbitMQ messages ready for delivery.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub messages_ready: Option<i64>,
+    /// RabbitMQ messages delivered but not yet acknowledged.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub messages_unacked: Option<i64>,
 }
 
 #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]

crates/dbx-core/src/schema.rs

@@ -3385,6 +3385,9 @@ for line in sys.stdin:
             internal: false,
             message_type: None,
             namespace: Some("default".to_string()),
+            message_count: None,
+            messages_ready: None,
+            messages_unacked: None,
         }]);
 
         assert_eq!(tables.len(), 1);