diff --git a/agents/drivers/rabbitmq/helpers.go b/agents/drivers/rabbitmq/helpers.go index c01784f61f..a9e5108b68 100644 --- a/agents/drivers/rabbitmq/helpers.go +++ b/agents/drivers/rabbitmq/helpers.go @@ -214,23 +214,31 @@ func floatOrNull(object jsonObject, key string) *float64 { } func boolOrDefault(object jsonObject, key string, fallback bool) bool { - if object == nil { + value := boolOrNull(object, key) + if value == nil { return fallback } + return *value +} + +func boolOrNull(object jsonObject, key string) *bool { + if object == nil { + return nil + } value, exists := object[key] if !exists || value == nil { - return fallback + return nil } switch typed := value.(type) { case bool: - return typed + return &typed case string: parsed, err := strconv.ParseBool(typed) if err == nil { - return parsed + return &parsed } } - return fallback + return nil } func integerProperty(properties jsonObject, key string) (int, bool) { diff --git a/agents/drivers/rabbitmq/mapping_test.go b/agents/drivers/rabbitmq/mapping_test.go index 6f31ac15c8..91e0045d3c 100644 --- a/agents/drivers/rabbitmq/mapping_test.go +++ b/agents/drivers/rabbitmq/mapping_test.go @@ -1,6 +1,9 @@ package main -import "testing" +import ( + "encoding/json" + "testing" +) func TestConsumersFromQueueInfo(t *testing.T) { info := mustObject(t, `{ @@ -126,6 +129,9 @@ func TestTopicInfoMappingIncludesQueueMessageCounts(t *testing.T) { if topic["messages"] != int64(12) || topic["messagesReady"] != int64(10) || topic["messagesUnacked"] != int64(2) { t.Fatalf("unexpected topic counts %#v", topic) } + if topic["consumers"] != int64(3) { + t.Fatalf("expected consumers wire field, got %#v", topic) + } minimal := topicInfoFromJSON(mustObject(t, `{"name":"empty"}`)) if _, ok := minimal["messagesReady"]; ok { @@ -143,3 +149,196 @@ func TestAttachVhost(t *testing.T) { t.Fatalf("unexpected info %#v", info) } } + +func TestTopicInfoMappingIncludesQueueFeaturesAndArguments(t *testing.T) { + topic := topicInfoFromJSON(mustObject(t, `{ + "name": "orders", + "durable": true, + "auto_delete": false, + "exclusive": true, + "state": "running", + "type": "quorum", + "messages": 123, + "messages_ready": 100, + "messages_unacknowledged": 23, + "consumers": 4, + "arguments": { + "x-queue-type": "quorum", + "x-message-ttl": 60000, + "x-max-length": 1000, + "x-queue-master-locator": "random" + } + }`)) + + if topic["durable"] != true || topic["autoDelete"] != false || topic["exclusive"] != true { + t.Fatalf("unexpected features %#v", topic) + } + if topic["state"] != "running" || topic["queueType"] != "quorum" { + t.Fatalf("unexpected state/type %#v", topic) + } + if topic["messages"] != int64(123) || topic["messagesReady"] != int64(100) || + topic["messagesUnacked"] != int64(23) || topic["consumers"] != int64(4) { + t.Fatalf("unexpected counts %#v", topic) + } + arguments := topic["arguments"].(jsonObject) + // Argument values must keep their real JSON types, not be stringified: + // numbers survive as json.Number and booleans/strings pass through. + if ttl, ok := arguments["x-message-ttl"].(json.Number); !ok || ttl.String() != "60000" { + t.Fatalf("unexpected x-message-ttl %#v", arguments["x-message-ttl"]) + } + if arguments["x-queue-type"] != "quorum" || arguments["x-queue-master-locator"] != "random" { + t.Fatalf("unexpected arguments %#v", arguments) + } +} + +func TestTopicInfoMappingFallsBackToXQueueTypeArgument(t *testing.T) { + // Older RabbitMQ versions do not report the explicit `type` field; the + // queue type then lives only in the x-queue-type argument. + topic := topicInfoFromJSON(mustObject(t, `{ + "name": "stream-q", + "arguments": {"x-queue-type": "stream"} + }`)) + if topic["queueType"] != "stream" { + t.Fatalf("expected x-queue-type fallback, got %#v", topic["queueType"]) + } + + minimal := topicInfoFromJSON(mustObject(t, `{"name": "plain"}`)) + if _, exists := minimal["queueType"]; exists { + t.Fatalf("unexpected queueType %#v", minimal) + } + if _, exists := minimal["arguments"]; exists { + t.Fatalf("unexpected arguments %#v", minimal) + } + if _, exists := minimal["exclusive"]; exists { + t.Fatalf("unexpected exclusive %#v", minimal) + } +} + +func TestTopicInfoMappingIncludesMessageStatsRates(t *testing.T) { + topic := topicInfoFromJSON(mustObject(t, `{ + "name": "orders", + "message_stats": { + "publish": 1000, + "publish_details": {"rate": 12.5}, + "deliver_get": 900, + "deliver_get_details": {"rate": 11.8}, + "ack": 880, + "ack_details": {"rate": 11.2} + } + }`)) + if topic["publishRate"] != 12.5 || topic["deliverRate"] != 11.8 || topic["ackRate"] != 11.2 { + t.Fatalf("unexpected rates %#v", topic) + } + + // No message_stats: rates must be absent, not fabricated as zero. + withoutStats := topicInfoFromJSON(mustObject(t, `{"name": "idle", "messages": 0}`)) + for _, key := range []string{"publishRate", "deliverRate", "ackRate"} { + if _, exists := withoutStats[key]; exists { + t.Fatalf("unexpected %s in %#v", key, withoutStats) + } + } + + // A real sampled rate of zero is preserved as zero (distinct from absent). + zeroRate := topicInfoFromJSON(mustObject(t, `{ + "name": "quiet", + "message_stats": {"publish_details": {"rate": 0}, "deliver_get_details": {"rate": 0}, "ack_details": {"rate": 0}} + }`)) + if zeroRate["publishRate"] != 0.0 || zeroRate["deliverRate"] != 0.0 || zeroRate["ackRate"] != 0.0 { + t.Fatalf("expected preserved zero rates, got %#v", zeroRate) + } +} + +func TestGetTopicStatsPreservesMessageStatsFromManagementPayload(t *testing.T) { + queue := mustObject(t, `{ + "name": "orders", + "messages": 123, + "messages_ready": 100, + "messages_unacknowledged": 23, + "consumers": 4, + "message_stats": { + "publish": 1000, + "publish_details": {"rate": 12.5}, + "deliver_get": 900, + "deliver_get_details": {"rate": 11.8}, + "ack": 880, + "ack_details": {"rate": 11.2} + } + }`) + // Mirror the Management API branch of getTopicStats: counts + rates all + // flow through, and a queue without message_stats omits every rate key. + messages := longOrDefault(queue, "messages", 0) + result := jsonObject{ + "name": "orders", + "messageCount": messages, + "consumerCount": longOrDefault(queue, "consumers", 0), + "totalMessages": messages, + } + putIfPresent(result, "messagesReady", longOrNull(queue, "messages_ready")) + putIfPresent(result, "messagesUnacked", longOrNull(queue, "messages_unacknowledged")) + if stats := objectOrNil(queue, "message_stats"); stats != nil { + putIfPresent(result, "publishRate", rateFromDetails(stats, "publish_details")) + putIfPresent(result, "deliverRate", rateFromDetails(stats, "deliver_get_details")) + putIfPresent(result, "ackRate", rateFromDetails(stats, "ack_details")) + putIfPresent(result, "publishTotal", longOrNull(stats, "publish")) + putIfPresent(result, "deliverGetTotal", longOrNull(stats, "deliver_get")) + putIfPresent(result, "ackTotal", longOrNull(stats, "ack")) + } + + if result["messageCount"] != int64(123) || result["messagesReady"] != int64(100) || + result["messagesUnacked"] != int64(23) || result["consumerCount"] != int64(4) { + t.Fatalf("unexpected counts %#v", result) + } + if result["publishRate"] != 12.5 || result["deliverRate"] != 11.8 || result["ackRate"] != 11.2 { + t.Fatalf("unexpected rates %#v", result) + } + if result["publishTotal"] != int64(1000) || result["deliverGetTotal"] != int64(900) || result["ackTotal"] != int64(880) { + t.Fatalf("unexpected totals %#v", result) + } + + withoutStats := jsonObject{"name": "idle", "messageCount": 0, "consumerCount": 0, "totalMessages": 0} + for _, key := range []string{"publishRate", "deliverRate", "ackRate", "publishTotal", "deliverGetTotal", "ackTotal"} { + if _, exists := withoutStats[key]; exists { + t.Fatalf("unexpected %s in %#v", key, withoutStats) + } + } +} + +func TestGetTopicConfigPreservesArgumentTypes(t *testing.T) { + configs := jsonObject{} + object := mustObject(t, `{ + "durable": true, + "auto_delete": false, + "exclusive": false, + "type": "quorum", + "arguments": { + "x-message-ttl": 60000, + "x-max-priority": 5, + "x-queue-type": "quorum", + "x-dead-letter-exchange": "dlx", + "x-single-active-consumer": true + } + }`) + configs["durable"] = boolOrDefault(object, "durable", false) + configs["auto_delete"] = boolOrDefault(object, "auto_delete", false) + configs["exclusive"] = boolOrDefault(object, "exclusive", false) + if queueType := queueTypeFromQueue(object); queueType != "" { + configs["queue_type"] = queueType + } + if arguments := objectOrNil(object, "arguments"); arguments != nil { + for key, value := range arguments { + configs[key] = value + } + } + + if configs["queue_type"] != "quorum" || configs["durable"] != true { + t.Fatalf("unexpected config %#v", configs) + } + // Numbers and booleans keep their types instead of fmt.Sprint strings: + // the decoder produced json.Number, and pass-through preserves it verbatim. + if ttl, ok := configs["x-message-ttl"].(json.Number); !ok || ttl.String() != "60000" { + t.Fatalf("unexpected x-message-ttl type %#v", configs["x-message-ttl"]) + } + if configs["x-single-active-consumer"] != true || configs["x-dead-letter-exchange"] != "dlx" { + t.Fatalf("unexpected argument values %#v", configs) + } +} diff --git a/agents/drivers/rabbitmq/operations.go b/agents/drivers/rabbitmq/operations.go index df60510f24..fdcdf75482 100644 --- a/agents/drivers/rabbitmq/operations.go +++ b/agents/drivers/rabbitmq/operations.go @@ -63,9 +63,36 @@ func topicInfoFromJSON(queue jsonObject) jsonObject { } putIfPresent(info, "messagesReady", longOrNull(queue, "messages_ready")) putIfPresent(info, "messagesUnacked", longOrNull(queue, "messages_unacknowledged")) + putIfPresent(info, "exclusive", boolOrNull(queue, "exclusive")) + if queueType := queueTypeFromQueue(queue); queueType != "" { + info["queueType"] = queueType + } + if arguments := objectOrNil(queue, "arguments"); arguments != nil && len(arguments) > 0 { + // Preserve original JSON types (numbers, booleans, nested values); the + // management API already decoded them, so pass them through verbatim. + info["arguments"] = arguments + } + if stats := objectOrNil(queue, "message_stats"); stats != nil { + putIfPresent(info, "publishRate", rateFromDetails(stats, "publish_details")) + putIfPresent(info, "deliverRate", rateFromDetails(stats, "deliver_get_details")) + putIfPresent(info, "ackRate", rateFromDetails(stats, "ack_details")) + } return info } +// queueTypeFromQueue prefers the management API's explicit `type` field +// (classic/quorum/stream, present on newer RabbitMQ versions) and falls back +// to the x-queue-type argument for older versions that only report it there. +func queueTypeFromQueue(queue jsonObject) string { + if explicit := stringOrEmpty(queue, "type"); explicit != "" { + return explicit + } + if arguments := objectOrNil(queue, "arguments"); arguments != nil { + return stringOrEmpty(arguments, "x-queue-type") + } + return "" +} + func (s *server) createTopic(params jsonObject) (any, error) { channel, err := s.channelFor(params) if err != nil { @@ -116,13 +143,29 @@ func (s *server) getTopicStats(params jsonObject) (any, error) { "/api/queues/"+urlEncodeVhost(vhost)+"/"+urlEncodePathSegment(name)) if managementError == nil { if info, ok := queue.(map[string]any); ok { - messages := longOrDefault(jsonObject(info), "messages", 0) - return jsonObject{ + object := jsonObject(info) + messages := longOrDefault(object, "messages", 0) + result := jsonObject{ "name": name, "messageCount": messages, - "consumerCount": longOrDefault(jsonObject(info), "consumers", 0), + "consumerCount": longOrDefault(object, "consumers", 0), "totalMessages": messages, - }, nil + } + putIfPresent(result, "messagesReady", longOrNull(object, "messages_ready")) + putIfPresent(result, "messagesUnacked", longOrNull(object, "messages_unacknowledged")) + if stats := objectOrNil(object, "message_stats"); stats != nil { + // Rates are only emitted when the management API actually + // sampled them; a missing entry stays absent so callers can + // distinguish "no data" from a real rate of zero. + putIfPresent(result, "publishRate", rateFromDetails(stats, "publish_details")) + putIfPresent(result, "deliverRate", rateFromDetails(stats, "deliver_get_details")) + putIfPresent(result, "ackRate", rateFromDetails(stats, "ack_details")) + // Cumulative counters, matching the management UI's totals. + putIfPresent(result, "publishTotal", longOrNull(stats, "publish")) + putIfPresent(result, "deliverGetTotal", longOrNull(stats, "deliver_get")) + putIfPresent(result, "ackTotal", longOrNull(stats, "ack")) + } + return result, nil } } else { fmt.Fprintln(os.Stderr, "Management API unavailable for queue stats, falling back to passive declare: "+managementError.Error()) @@ -164,13 +207,15 @@ func (s *server) getTopicConfig(params jsonObject) (any, error) { configs["durable"] = boolOrDefault(object, "durable", false) configs["auto_delete"] = boolOrDefault(object, "auto_delete", false) configs["exclusive"] = boolOrDefault(object, "exclusive", false) + if queueType := queueTypeFromQueue(object); queueType != "" { + configs["queue_type"] = queueType + } if arguments := objectOrNil(object, "arguments"); arguments != nil { + // Preserve the real JSON types the management API returned + // (numbers, booleans, nested values) instead of stringifying + // every argument, so config round-trips and display stay true. for key, value := range arguments { - if value == nil { - configs[key] = nil - } else { - configs[key] = fmt.Sprint(value) - } + configs[key] = value } } } @@ -1345,6 +1390,10 @@ func putIfPresent(info jsonObject, key string, value any) { if typed != nil { info[key] = *typed } + case *bool: + if typed != nil { + info[key] = *typed + } case nil: default: info[key] = typed diff --git a/apps/desktop/src/components/mq/MonitoringPanel.vue b/apps/desktop/src/components/mq/MonitoringPanel.vue index 4a89c8b2a5..5e6b185cbe 100644 --- a/apps/desktop/src/components/mq/MonitoringPanel.vue +++ b/apps/desktop/src/components/mq/MonitoringPanel.vue @@ -25,8 +25,8 @@ interface Props { interface MetricPoint { time: string; - msgRateIn: number; - msgRateOut: number; + msgRateIn: number | null; + msgRateOut: number | null; backlogSize: number; msgBacklog: number; consumerLagMs: number; @@ -72,6 +72,16 @@ const selectedPartitionName = ref(); let refreshTimer: number | undefined; const history = ref([]); const MAX_HISTORY_POINTS = 60; +const messageFlowStatus = computed(() => { + if (!stats.value || stats.value.ratesUnavailable) { + return { kind: "unavailable", label: "-" }; + } + const active = stats.value.msgRateIn > 0 || stats.value.msgRateOut > 0; + return { + kind: active ? "healthy" : "idle", + label: active ? t("mqMonitoring.flowActive") : t("mqMonitoring.flowIdle"), + }; +}); const partitionRows = computed(() => extractPartitionRows(stats.value?.raw)); const kafkaPartitionRows = computed(() => extractKafkaPartitionRows(stats.value?.raw)); @@ -213,10 +223,11 @@ function refreshNow() { } function appendHistoryPoint(statsData: TopicStats, backlogData?: BacklogStats) { + const ratesUnavailable = statsData.ratesUnavailable === true; const point: MetricPoint = { time: new Date().toLocaleTimeString(), - msgRateIn: statsData.msgRateIn, - msgRateOut: statsData.msgRateOut, + msgRateIn: ratesUnavailable ? null : statsData.msgRateIn, + msgRateOut: ratesUnavailable ? null : statsData.msgRateOut, backlogSize: statsData.backlogSize, msgBacklog: backlogData?.msgBacklog ?? 0, consumerLagMs: extractConsumerLagMs(statsData.raw), @@ -351,6 +362,10 @@ function formatNumber(num: number): string { return num.toLocaleString(); } +function formatMessageRate(rate: number, unavailable?: boolean): string { + return unavailable ? "-" : `${rate.toFixed(2)} msg/s`; +} + watch( () => props.topic, () => { @@ -615,14 +630,14 @@ onUnmounted(() => {
{{ t("mqMonitoring.inboundRate") }}
-
{{ stats.msgRateIn.toFixed(2) }} msg/s
+
{{ formatMessageRate(stats.msgRateIn, stats.ratesUnavailable) }}
{{ t("mqMonitoring.outboundRate") }}
-
{{ stats.msgRateOut.toFixed(2) }} msg/s
+
{{ formatMessageRate(stats.msgRateOut, stats.ratesUnavailable) }}
@@ -645,7 +660,7 @@ onUnmounted(() => {

{{ t("mqMonitoring.rateTrend") }}

- +

{{ t("mqMonitoring.backlogTrend") }}

@@ -820,8 +835,8 @@ onUnmounted(() => {
{{ t("mqMonitoring.messageFlow") }}: - - {{ stats.msgRateIn > 0 || stats.msgRateOut > 0 ? t("mqMonitoring.flowActive") : t("mqMonitoring.flowIdle") }} + + {{ messageFlowStatus.label }}
@@ -1467,6 +1482,11 @@ onUnmounted(() => { color: var(--monitor-muted); } +.health-badge.unavailable { + background: color-mix(in srgb, var(--monitor-muted) 8%, transparent); + color: var(--monitor-muted); +} + .monitoring-message-heading { display: flex; align-items: center; diff --git a/apps/desktop/src/components/mq/TopicsPanel.vue b/apps/desktop/src/components/mq/TopicsPanel.vue index c8bba83cca..6609388d3c 100644 --- a/apps/desktop/src/components/mq/TopicsPanel.vue +++ b/apps/desktop/src/components/mq/TopicsPanel.vue @@ -12,6 +12,7 @@ import ExchangesPanel from "./ExchangesPanel.vue"; import MqTypeFilterBar from "./shared/MqTypeFilterBar.vue"; import type { MqTab } from "@/lib/mq/mqConsoleDefaults"; import { isAllVhostsNamespace, resolveMqRowNamespace } from "@/lib/mq/mqConsoleDefaults"; +import { formatMqRateCompact, rabbitMqArgumentsSummary, rabbitMqFeatureBadges, rabbitMqQueueType, type RabbitMqFeatureBadge } from "@/lib/mq/rabbitmqQueueFeatures"; import { formatError } from "@/lib/backend/errorUtils"; import { DEFAULT_ROCKETMQ_TOPIC_TYPE_FILTERS, isProtectedRocketMqTopic, isRocketMqBusinessMessageType, matchesRocketMqTypeFilters, resolveRocketMqMessageType, ROCKETMQ_CREATABLE_TOPIC_MESSAGE_TYPES, ROCKETMQ_TOPIC_MESSAGE_TYPES } from "@/lib/mq/rocketmqTopicTypes"; import { useMqMutationGuard } from "@/composables/useMqMutationGuard"; @@ -141,22 +142,35 @@ const virtualTopicRows = computed(() => const topicsGridTemplate = computed(() => { const cols: string[] = ["minmax(180px, 1.6fr)"]; if (showNamespaceColumn.value) cols.push("minmax(100px, 0.8fr)"); - cols.push("120px"); - if (!isRocketMqCluster.value) cols.push("140px"); + if (isRabbitMqCluster.value) { + // RabbitMQ queues: type(+state) | features | messages | consumers | rates | actions + cols.push("110px", "170px", "90px", "70px", "190px"); + } else { + cols.push("120px"); + } + if (!isRocketMqCluster.value && !isRabbitMqCluster.value) cols.push("140px"); // RocketMQ keeps tiled row actions (status/route/consumers/…) — reserve a wider actions column. - cols.push(isRocketMqCluster.value ? "minmax(560px, 2.2fr)" : "minmax(200px, 1fr)"); + cols.push(isRocketMqCluster.value ? "minmax(560px, 2.2fr)" : "minmax(150px, 1fr)"); return cols.join(" "); }); /** Min content width from grid track mins + gaps + padding; enables shared horizontal scroll. */ const topicsTableMinWidthPx = computed(() => { - let min = 180 + 120; // name + type + let min = 180; // name if (showNamespaceColumn.value) min += 100; - if (!isRocketMqCluster.value) min += 140; - min += isRocketMqCluster.value ? 560 : 200; - let colCount = 3; + let colCount = 0; + if (isRabbitMqCluster.value) { + // type + features + messages + consumers + rates + actions + min += 110 + 170 + 90 + 70 + 190 + 150; + colCount = 7; // name, type, features, messages, consumers, rates, actions + } else { + min += 120; // type + if (!isRocketMqCluster.value) min += 140; // partitions + min += isRocketMqCluster.value ? 560 : 200; // actions + // RocketMQ: name + type + actions. Others: + partitions. + colCount = isRocketMqCluster.value ? 3 : 4; + } if (showNamespaceColumn.value) colCount += 1; - if (!isRocketMqCluster.value) colCount += 1; min += (colCount - 1) * 8; // column-gap min += 24; // horizontal padding return min; @@ -177,6 +191,37 @@ function rabbitMqReadyMessageCount(topic: TopicInfo): number { return topic.messagesReady ?? topic.messageCount ?? 0; } +/** Queue type label for the type column; unknown types are never guessed. */ +function rabbitMqQueueTypeLabel(topic: TopicInfo): string { + return rabbitMqQueueType(topic) ?? t("mqTopics.rabbitmqQueueTypeUnknown"); +} + +function rabbitMqQueueTypeBadgeClass(topic: TopicInfo): string { + const type = rabbitMqQueueType(topic); + if (type === "quorum") return "badge-info"; + if (type === "stream") return "badge-warning"; + return "badge-default"; +} + +function rabbitMqFeatureBadgesFor(topic: TopicInfo): RabbitMqFeatureBadge[] { + return rabbitMqFeatureBadges(topic); +} + +/** Tooltip for the message-count column: Ready / Unacked / Total / Consumers. */ +function rabbitMqMessageCountTitle(topic: TopicInfo): string { + const parts = [`${t("mqRabbitMqMonitoring.messagesReady")}: ${topic.messagesReady ?? "-"}`, `${t("mqRabbitMqMonitoring.messagesUnacked")}: ${topic.messagesUnacked ?? "-"}`, `${t("mqTopics.messageCount")}: ${topic.messageCount ?? "-"}`]; + if (topic.consumerCount !== undefined) parts.push(`${t("mqTopics.consumers")}: ${topic.consumerCount}`); + if (topic.state) parts.push(`${t("mqTopics.rabbitmqState")}: ${topic.state}`); + const summary = rabbitMqArgumentsSummary(topic); + if (summary) parts.push(`${t("mqTopics.rabbitmqArguments")}: ${summary}`); + return parts.join(" · "); +} + +/** Tooltip for the rates column; rates are msg/s sampled by the management API. */ +function rabbitMqRatesTitle(): string { + return `${t("mqTopics.rabbitmqPublishRate")} / ${t("mqTopics.rabbitmqDeliverRate")} / ${t("mqTopics.rabbitmqAckRate")} (msg/s)`; +} + function toggleRabbitMqMessageSort() { rabbitMqMessageSort.value = rabbitMqMessageSort.value === "desc" ? "asc" : "desc"; } @@ -551,7 +596,9 @@ watch(newPartitions, () => {
{{ t("mqTopics.name") }}
{{ t("mqAdmin.namespace") }}
-
{{ t("mqTopics.type") }}
+
{{ t("mqTopics.rabbitmqQueueType") }}
+
{{ t("mqTopics.type") }}
+
{{ t("mqTopics.rabbitmqFeatures") }}
{{ t("mqTopics.partitions") }}
+
{{ t("mqTopics.consumers") }}
+
{{ t("mqTopics.rabbitmqRates") }}
{{ t("mqTopics.actions") }}
@@ -571,18 +620,40 @@ watch(newPartitions, () => {
{{ row.topic.namespace || "-" }}
-
+
+ + {{ rabbitMqQueueTypeLabel(row.topic) }} + + {{ row.topic.state }} +
+
{{ topicTypeLabel(row.topic) }}
-
+
+ + + {{ badge.label }} + + + - +
+
{{ rabbitMqReadyMessageCount(row.topic) }}
{{ row.topic.partitions ? t("mqTopics.partitionCount", { count: row.topic.partitions }) : t("mqTopics.partitionsUnknown") }} -
+
{{ row.topic.consumerCount ?? "-" }}
+
+ {{ formatMqRateCompact(row.topic.publishRate) }} + / + {{ formatMqRateCompact(row.topic.deliverRate) }} + / + {{ formatMqRateCompact(row.topic.ackRate) }} +