From 1259c34db85bd1f86f879f1d4a72dc6ccec3c566 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Wed, 12 Aug 2026 14:00:36 +0800 Subject: [PATCH] feat(rabbitmq): show sortable queue message counts Closes #4286 --- agents/drivers/rabbitmq/mapping_test.go | 19 +++ agents/drivers/rabbitmq/operations.go | 23 ++-- .../desktop/src/components/mq/TopicsPanel.vue | 45 ++++++- .../mq/__tests__/TopicsPanel.spec.ts | 124 ++++++++++++++++++ apps/desktop/src/types/mq.ts | 6 + crates/dbx-core/src/mq/adapters/kafka.rs | 3 + crates/dbx-core/src/mq/adapters/pulsar.rs | 6 + crates/dbx-core/src/mq/adapters/rabbitmq.rs | 53 +++++--- crates/dbx-core/src/mq/adapters/rocketmq.rs | 3 + crates/dbx-core/src/mq/types.rs | 9 ++ crates/dbx-core/src/schema.rs | 3 + 11 files changed, 266 insertions(+), 28 deletions(-) create mode 100644 apps/desktop/src/components/mq/__tests__/TopicsPanel.spec.ts diff --git a/agents/drivers/rabbitmq/mapping_test.go b/agents/drivers/rabbitmq/mapping_test.go index 61885a952c..6f31ac15c8 100644 --- a/agents/drivers/rabbitmq/mapping_test.go +++ b/agents/drivers/rabbitmq/mapping_test.go @@ -117,6 +117,25 @@ func TestOverviewAndNodeMappings(t *testing.T) { } } +func TestTopicInfoMappingIncludesQueueMessageCounts(t *testing.T) { + topic := topicInfoFromJSON(mustObject(t, `{ + "name":"orders","durable":true,"messages":12, + "messages_ready":10,"messages_unacknowledged":2,"consumers":3 + }`)) + + if topic["messages"] != int64(12) || topic["messagesReady"] != int64(10) || topic["messagesUnacked"] != int64(2) { + t.Fatalf("unexpected topic counts %#v", topic) + } + + minimal := topicInfoFromJSON(mustObject(t, `{"name":"empty"}`)) + if _, ok := minimal["messagesReady"]; ok { + t.Fatalf("unexpected ready count %#v", minimal) + } + if _, ok := minimal["messagesUnacked"]; ok { + t.Fatalf("unexpected unacked count %#v", minimal) + } +} + func TestAttachVhost(t *testing.T) { info := jsonObject{"name": "q"} attachVhost(info, mustObject(t, `{"vhost":"orders"}`)) diff --git a/agents/drivers/rabbitmq/operations.go b/agents/drivers/rabbitmq/operations.go index 51854f923e..df60510f24 100644 --- a/agents/drivers/rabbitmq/operations.go +++ b/agents/drivers/rabbitmq/operations.go @@ -40,14 +40,7 @@ func (s *server) listTopics(params jsonObject) (any, error) { if !ok { continue } - info := jsonObject{ - "name": stringOrEmpty(jsonObject(queue), "name"), - "durable": boolOrDefault(jsonObject(queue), "durable", false), - "autoDelete": boolOrDefault(jsonObject(queue), "auto_delete", false), - "state": stringOrEmpty(jsonObject(queue), "state"), - "messages": longOrDefault(jsonObject(queue), "messages", 0), - "consumers": longOrDefault(jsonObject(queue), "consumers", 0), - } + info := topicInfoFromJSON(jsonObject(queue)) if allVhosts { attachVhost(info, jsonObject(queue)) } @@ -59,6 +52,20 @@ func (s *server) listTopics(params jsonObject) (any, error) { return jsonObject{"topics": topics}, nil } +func topicInfoFromJSON(queue jsonObject) jsonObject { + info := jsonObject{ + "name": stringOrEmpty(queue, "name"), + "durable": boolOrDefault(queue, "durable", false), + "autoDelete": boolOrDefault(queue, "auto_delete", false), + "state": stringOrEmpty(queue, "state"), + "messages": longOrDefault(queue, "messages", 0), + "consumers": longOrDefault(queue, "consumers", 0), + } + putIfPresent(info, "messagesReady", longOrNull(queue, "messages_ready")) + putIfPresent(info, "messagesUnacked", longOrNull(queue, "messages_unacknowledged")) + return info +} + func (s *server) createTopic(params jsonObject) (any, error) { channel, err := s.channelFor(params) if err != nil { diff --git a/apps/desktop/src/components/mq/TopicsPanel.vue b/apps/desktop/src/components/mq/TopicsPanel.vue index 68dab4bc5c..c8bba83cca 100644 --- a/apps/desktop/src/components/mq/TopicsPanel.vue +++ b/apps/desktop/src/components/mq/TopicsPanel.vue @@ -96,6 +96,7 @@ const showNamespaceColumn = computed(() => isRabbitMqCluster.value && isAllVhost // RabbitMQ splits the topics tab into Queues (topics table) and Exchanges views. const showRabbitMqSubTabs = computed(() => isRabbitMqCluster.value && props.supportsExchanges === true); const rabbitMqSubTab = ref<"queues" | "exchanges">("queues"); +const rabbitMqMessageSort = ref<"asc" | "desc" | null>(null); const rocketMqTopicTypeOptions = ROCKETMQ_TOPIC_MESSAGE_TYPES; const rocketMqCreatableTopicTypes = ROCKETMQ_CREATABLE_TOPIC_MESSAGE_TYPES; const rocketMqClusterName = computed(() => clusterInfo.value?.clusterId ?? "-"); @@ -120,9 +121,18 @@ const filteredTopics = computed(() => { }); }); +const displayedTopics = computed(() => { + if (!isRabbitMqCluster.value || !rabbitMqMessageSort.value) return filteredTopics.value; + const direction = rabbitMqMessageSort.value === "asc" ? 1 : -1; + return [...filteredTopics.value].sort((left, right) => { + const countDifference = (rabbitMqReadyMessageCount(left) - rabbitMqReadyMessageCount(right)) * direction; + return countDifference || left.shortName.localeCompare(right.shortName); + }); +}); + /** Stable ids for RecycleScroller; avoids mounting all topic rows at once. */ const virtualTopicRows = computed(() => - filteredTopics.value.map((topic) => ({ + displayedTopics.value.map((topic) => ({ id: showNamespaceColumn.value ? `${topic.namespace ?? ""}:${topic.name}` : topic.name, topic, })), @@ -163,6 +173,14 @@ function topicRowSelected(topic: TopicInfo): boolean { return selectedTopic.value?.name === topic.name && selectedTopic.value?.namespace === topic.namespace; } +function rabbitMqReadyMessageCount(topic: TopicInfo): number { + return topic.messagesReady ?? topic.messageCount ?? 0; +} + +function toggleRabbitMqMessageSort() { + rabbitMqMessageSort.value = rabbitMqMessageSort.value === "desc" ? "asc" : "desc"; +} + function topicTypeLabel(topic: TopicInfo): string { if (isRocketMqCluster.value) { const type = resolveRocketMqMessageType(topic); @@ -534,7 +552,13 @@ watch(newPartitions, () => {
{{ t("mqTopics.name") }}
{{ t("mqAdmin.namespace") }}
{{ t("mqTopics.type") }}
-
{{ t("mqTopics.partitions") }}
+
+ +
+
{{ t("mqTopics.partitions") }}
{{ t("mqTopics.actions") }}
@@ -552,7 +576,10 @@ watch(newPartitions, () => { {{ topicTypeLabel(row.topic) }} -
+
+ {{ rabbitMqReadyMessageCount(row.topic) }} +
+
{{ row.topic.partitions ? t("mqTopics.partitionCount", { count: row.topic.partitions }) : t("mqTopics.partitionsUnknown") }} -
@@ -922,6 +949,18 @@ watch(newPartitions, () => { color: var(--color-text-secondary); } +.topics-sort-button { + display: inline-flex; + align-items: center; + gap: 4px; + padding: 0; + border: 0; + background: transparent; + color: inherit; + font: inherit; + cursor: pointer; +} + .topics-scroller { flex: 1; min-height: 0; diff --git a/apps/desktop/src/components/mq/__tests__/TopicsPanel.spec.ts b/apps/desktop/src/components/mq/__tests__/TopicsPanel.spec.ts new file mode 100644 index 0000000000..bfd0661996 --- /dev/null +++ b/apps/desktop/src/components/mq/__tests__/TopicsPanel.spec.ts @@ -0,0 +1,124 @@ +// @vitest-environment happy-dom + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createApp, defineComponent, h, nextTick, type App } from "vue"; + +const backend = vi.hoisted(() => ({ + mqListTopics: vi.fn(), + mqCreateTopic: vi.fn(), + mqDeleteTopic: vi.fn(), + mqUpdatePartitions: vi.fn(), + mqGetClusterInfo: vi.fn(), +})); + +vi.mock("vue-i18n", () => ({ + useI18n: () => ({ t: (key: string) => key }), +})); + +vi.mock("@/lib/backend/api", () => backend); + +vi.mock("vue-virtual-scroller", () => ({ + RecycleScroller: defineComponent({ + props: { items: { type: Array, default: () => [] } }, + setup(props, { slots }) { + return () => + h( + "div", + (props.items as unknown[]).flatMap((item) => slots.default?.({ item }) ?? []), + ); + }, + }), +})); + +vi.mock("@/composables/useMqMutationGuard", () => ({ + useMqMutationGuard: () => ({ confirmMqWrite: vi.fn().mockResolvedValue(true) }), +})); + +vi.mock("@/components/mq/rocketmq/RocketMqTopicDialogs.vue", () => ({ + default: defineComponent({ setup: () => () => h("div") }), +})); + +vi.mock("@/components/mq/SendMessagePanel.vue", () => ({ + default: defineComponent({ setup: () => () => h("div") }), +})); + +vi.mock("@/components/mq/ExchangesPanel.vue", () => ({ + default: defineComponent({ setup: () => () => h("div") }), +})); + +vi.mock("@/components/mq/shared/MqTypeFilterBar.vue", () => ({ + default: defineComponent({ + setup: + (_, { slots }) => + () => + h("div", slots.default?.()), + }), +})); + +vi.mock("@/components/editor/DangerConfirmDialog.vue", () => ({ + default: defineComponent({ setup: () => () => h("div") }), +})); + +import TopicsPanel from "@/components/mq/TopicsPanel.vue"; + +let app: App | null = null; +let root: HTMLDivElement | null = null; + +async function flushUi() { + await Promise.resolve(); + await Promise.resolve(); + await nextTick(); +} + +async function mountRabbitMqPanel() { + root = document.createElement("div"); + document.body.appendChild(root); + app = createApp(TopicsPanel, { + connectionId: "rabbit-1", + tenant: "_rabbitmq", + namespace: "/", + mqSystemKind: "rabbitmq", + }); + app.mount(root); + await flushUi(); + return root; +} + +function rowNames(container: ParentNode): string[] { + return [...container.querySelectorAll(".topic-name-text")].map((item) => item.textContent ?? ""); +} + +beforeEach(() => { + Object.values(backend).forEach((mock) => mock.mockReset()); + backend.mqListTopics.mockResolvedValue([ + { name: "alpha", shortName: "alpha", partitioned: false, persistent: true, messageCount: 4, messagesReady: 3, messagesUnacked: 1 }, + { name: "beta", shortName: "beta", partitioned: false, persistent: true, messageCount: 12, messagesReady: 12, messagesUnacked: 0 }, + ]); +}); + +afterEach(() => { + app?.unmount(); + app = null; + root?.remove(); + root = null; +}); + +describe("TopicsPanel RabbitMQ queue messages", () => { + it("shows ready message counts and toggles descending and ascending sorting", async () => { + const panel = await mountRabbitMqPanel(); + + expect([...panel.querySelectorAll('[data-testid="rabbitmq-message-count"]')].map((item) => item.textContent?.trim())).toEqual(["3", "12"]); + expect(rowNames(panel)).toEqual(["alpha", "beta"]); + + const sortButton = panel.querySelector('[data-testid="rabbitmq-message-sort"]'); + if (!sortButton) throw new Error("RabbitMQ message sort button not found"); + + sortButton.click(); + await nextTick(); + expect(rowNames(panel)).toEqual(["beta", "alpha"]); + + sortButton.click(); + await nextTick(); + expect(rowNames(panel)).toEqual(["alpha", "beta"]); + }); +}); diff --git a/apps/desktop/src/types/mq.ts b/apps/desktop/src/types/mq.ts index 3c16fc11c9..4412656d8f 100644 --- a/apps/desktop/src/types/mq.ts +++ b/apps/desktop/src/types/mq.ts @@ -156,6 +156,12 @@ export interface TopicInfo { messageType?: RocketMqTopicMessageType | string; /** RabbitMQ: owning virtual host, present when listing across all vhosts. */ namespace?: string; + /** RabbitMQ total queue messages, including ready and unacknowledged messages. */ + messageCount?: number; + /** RabbitMQ messages ready for delivery. */ + messagesReady?: number; + /** RabbitMQ messages delivered but not yet acknowledged. */ + messagesUnacked?: number; } export interface ListTopicsOpts { diff --git a/crates/dbx-core/src/mq/adapters/kafka.rs b/crates/dbx-core/src/mq/adapters/kafka.rs index 611c8922fa..2bef343f7c 100644 --- a/crates/dbx-core/src/mq/adapters/kafka.rs +++ b/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()) diff --git a/crates/dbx-core/src/mq/adapters/pulsar.rs b/crates/dbx-core/src/mq/adapters/pulsar.rs index 436ce0193f..c311f5b011 100644 --- a/crates/dbx-core/src/mq/adapters/pulsar.rs +++ b/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, }); } } diff --git a/crates/dbx-core/src/mq/adapters/rabbitmq.rs b/crates/dbx-core/src/mq/adapters/rabbitmq.rs index 8c7d783c87..956484a38e 100644 --- a/crates/dbx-core/src/mq/adapters/rabbitmq.rs +++ b/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) -> 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!({ diff --git a/crates/dbx-core/src/mq/adapters/rocketmq.rs b/crates/dbx-core/src/mq/adapters/rocketmq.rs index 46a1bf4c8f..48fdde5dff 100644 --- a/crates/dbx-core/src/mq/adapters/rocketmq.rs +++ b/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, } } diff --git a/crates/dbx-core/src/mq/types.rs b/crates/dbx-core/src/mq/types.rs index 11430d6f56..500f5aa538 100644 --- a/crates/dbx-core/src/mq/types.rs +++ b/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, + /// RabbitMQ total queue messages, including ready and unacknowledged messages. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub message_count: Option, + /// RabbitMQ messages ready for delivery. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub messages_ready: Option, + /// RabbitMQ messages delivered but not yet acknowledged. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub messages_unacked: Option, } #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)] diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 26c57014e4..429ed6ea27 100644 --- a/crates/dbx-core/src/schema.rs +++ b/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);