✨ feat(rocketmq): 新增 RocketMQ 数据源连接与测试发消息支持

internal/app/db_context.go

@@ -16,6 +16,8 @@ func normalizeRunConfig(config connection.ConnectionConfig, dbName string) conne
 	}
 
 	switch strings.ToLower(strings.TrimSpace(config.Type)) {
+	case "rocketmq", "rocket-mq", "rocket_mq", "apache-rocketmq", "apache_rocketmq", "rmq":
+		// RocketMQ 的 Database 字段表示默认 Topic，不能被树上的 synthetic database(topics) 覆盖。
 	case "mqtt", "mqtts":
 		// MQTT 的 Database 字段表示默认 Topic，不能被树上的 synthetic database(topics) 覆盖。
 	case "kafka", "apache-kafka", "apache_kafka":
@@ -55,7 +57,7 @@ func normalizeSchemaAndTable(config connection.ConnectionConfig, dbName string,
 
 	// Elasticsearch：索引名可能含多个点（如 iot_pro_biz_operate_log.index.20240626），
 	// 不能按点分割，直接返回原始数据库名和完整表名。
-	if dbType == "elasticsearch" || dbType == "iotdb" || dbType == "mqtt" || dbType == "kafka" || dbType == "rabbitmq" {
+	if dbType == "elasticsearch" || dbType == "iotdb" || dbType == "rocketmq" || dbType == "mqtt" || dbType == "kafka" || dbType == "rabbitmq" {
 		return rawDB, rawTable
 	}
 
@@ -114,7 +116,7 @@ func normalizeSchemaAndTable(config connection.ConnectionConfig, dbName string,
 func normalizeMetadataSchemaAndTable(config connection.ConnectionConfig, dbName string, tableName string) (string, string) {
 	schema, table := normalizeSchemaAndTable(config, dbName, tableName)
 	switch resolveDDLDBType(config) {
-	case "mqtt", "kafka", "rabbitmq":
+	case "rocketmq", "mqtt", "kafka", "rabbitmq":
 		return schema, table
 	case "postgres", "kingbase", "highgo", "vastbase", "opengauss", "gaussdb":
 		rawTable := strings.TrimSpace(tableName)

internal/app/db_context_test.go

@@ -344,6 +344,18 @@ func TestNormalizeSchemaAndTable_MQTTPreservesTopicFilter(t *testing.T) {
 	}
 }
 
+func TestNormalizeSchemaAndTable_RocketMQPreservesTopicName(t *testing.T) {
+	t.Parallel()
+
+	schemaOrDb, table := normalizeSchemaAndTable(connection.ConnectionConfig{
+		Type: "rocketmq",
+	}, "topics", "orders.events.v1")
+
+	if schemaOrDb != "topics" || table != "orders.events.v1" {
+		t.Fatalf("expected rocketmq topic name to stay intact, got %q.%q", schemaOrDb, table)
+	}
+}
+
 func TestNormalizeSchemaAndTable_RabbitMQPreservesDottedQueueName(t *testing.T) {
 	t.Parallel()
 
@@ -380,6 +392,18 @@ func TestNormalizeMetadataSchemaAndTable_MQTTPreservesTopicFilter(t *testing.T)
 	}
 }
 
+func TestNormalizeMetadataSchemaAndTable_RocketMQPreservesTopicName(t *testing.T) {
+	t.Parallel()
+
+	schemaOrDb, table := normalizeMetadataSchemaAndTable(connection.ConnectionConfig{
+		Type: "rocketmq",
+	}, "topics", "orders.events.v1")
+
+	if schemaOrDb != "topics" || table != "orders.events.v1" {
+		t.Fatalf("expected rocketmq metadata topic name to stay intact, got %q.%q", schemaOrDb, table)
+	}
+}
+
 func TestNormalizeMetadataSchemaAndTable_RabbitMQPreservesDottedQueueName(t *testing.T) {
 	t.Parallel()
 

internal/app/methods_db.go

@@ -315,8 +315,8 @@ func normalizeSchemaAndTableByType(dbType string, dbName string, tableName strin
 		return rawDB, rawTable
 	}
 
-	// Elasticsearch / MQTT / RabbitMQ / Kafka：对象名可能含多个点或路径，不能按点分割
-	if dbType == "elasticsearch" || dbType == "mqtt" || dbType == "kafka" || dbType == "rabbitmq" {
+	// Elasticsearch / RocketMQ / MQTT / RabbitMQ / Kafka：对象名可能含多个点或路径，不能按点分割
+	if dbType == "elasticsearch" || dbType == "rocketmq" || dbType == "mqtt" || dbType == "kafka" || dbType == "rabbitmq" {
 		return rawDB, rawTable
 	}
 

internal/app/methods_db_create_statement_test.go

@@ -175,6 +175,15 @@ func TestNormalizeSchemaAndTableByType_MQTTPreservesTopicFilter(t *testing.T) {
 	}
 }
 
+func TestNormalizeSchemaAndTableByType_RocketMQPreservesTopicName(t *testing.T) {
+	t.Parallel()
+
+	schema, table := normalizeSchemaAndTableByType("rocketmq", "topics", "orders.events.v1")
+	if schema != "topics" || table != "orders.events.v1" {
+		t.Fatalf("expected rocketmq topic name to stay intact, got %q.%q", schema, table)
+	}
+}
+
 func TestNormalizeSchemaAndTableByType_RabbitMQPreservesDottedQueueName(t *testing.T) {
 	t.Parallel()
 

internal/app/methods_driver.go

@@ -1432,6 +1432,8 @@ func normalizeDriverType(driverType string) string {
 		return "gaussdb"
 	case "goldendb", "greatdb", "gdb":
 		return "goldendb"
+	case "rocketmq", "rocket-mq", "rocket_mq", "apache-rocketmq", "apache_rocketmq", "rmq":
+		return "rocketmq"
 	case "mqtt", "mqtts":
 		return "mqtt"
 	case "kafka", "apache-kafka", "apache_kafka":
@@ -1507,6 +1509,7 @@ func allDriverDefinitionsWithPackages(packages map[string]pinnedDriverPackage) [
 		{Type: "oracle", Name: "Oracle", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "redis", Name: "Redis", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "postgres", Name: "PostgreSQL", Engine: driverEngineGo, BuiltIn: true},
+		{Type: "rocketmq", Name: "RocketMQ", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "mqtt", Name: "MQTT", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "kafka", Name: "Kafka", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "rabbitmq", Name: "RabbitMQ", Engine: driverEngineGo, BuiltIn: true},

internal/app/methods_driver_version_test.go

@@ -534,6 +534,22 @@ func TestMQTTDriverDefinitionIsBuiltIn(t *testing.T) {
 	}
 }
 
+func TestRocketMQDriverDefinitionIsBuiltIn(t *testing.T) {
+	definition, ok := resolveDriverDefinition("rmq")
+	if !ok {
+		t.Fatal("expected rocketmq driver definition")
+	}
+	if definition.Name != "RocketMQ" {
+		t.Fatalf("unexpected rocketmq driver name: %q", definition.Name)
+	}
+	if !definition.BuiltIn {
+		t.Fatal("expected rocketmq to be a built-in driver")
+	}
+	if definition.PinnedVersion != "" || definition.DefaultDownloadURL != "" {
+		t.Fatalf("expected rocketmq builtin definition to omit optional-agent metadata: %#v", definition)
+	}
+}
+
 func TestRabbitMQDriverDefinitionIsBuiltIn(t *testing.T) {
 	definition, ok := resolveDriverDefinition("rabbit-mq")
 	if !ok {

internal/db/database.go

@@ -489,6 +489,9 @@ var databaseFactories = map[string]databaseFactory{
 	"qdrant": func() Database {
 		return &QdrantDB{}
 	},
+	"rocketmq": func() Database {
+		return &RocketMQDB{}
+	},
 	"mqtt": func() Database {
 		return &MQTTDB{}
 	},
@@ -538,6 +541,8 @@ func normalizeDatabaseType(dbType string) string {
 		return "chroma"
 	case "qdrantdb", "qdrant-db":
 		return "qdrant"
+	case "rocketmq", "rocket-mq", "rocket_mq", "apache-rocketmq", "apache_rocketmq", "rmq":
+		return "rocketmq"
 	case "mqtt", "mqtts":
 		return "mqtt"
 	case "kafka", "apache-kafka", "apache_kafka":

internal/db/driver_support.go

@@ -19,6 +19,7 @@ var coreBuiltinDrivers = map[string]struct{}{
 	"postgres": {},
 	"chroma":   {},
 	"qdrant":   {},
+	"rocketmq": {},
 	"mqtt":     {},
 	"kafka":    {},
 	"rabbitmq": {},
@@ -82,6 +83,8 @@ func normalizeRuntimeDriverType(driverType string) string {
 		return "chroma"
 	case "qdrantdb", "qdrant-db":
 		return "qdrant"
+	case "rocketmq", "rocket-mq", "rocket_mq", "apache-rocketmq", "apache_rocketmq", "rmq":
+		return "rocketmq"
 	case "mqtt", "mqtts":
 		return "mqtt"
 	case "apache-iotdb", "apache_iotdb", "iotdb":
@@ -151,6 +154,8 @@ func driverDisplayName(driverType string) string {
 		return "Chroma"
 	case "qdrant":
 		return "Qdrant"
+	case "rocketmq":
+		return "RocketMQ"
 	case "mqtt":
 		return "MQTT"
 	case "kafka":

internal/db/rocketmq_impl_test.go

@@ -0,0 +1,312 @@
+package db
+
+import (
+	"context"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+
+	rocketmqprimitive "github.com/apache/rocketmq-client-go/v2/primitive"
+)
+
+type fakeRocketMQRuntime struct {
+	listTopicsResult   []rocketmqTopicInfo
+	describeResult     rocketmqTopicDescription
+	fetchResult        []rocketmqMessageRecord
+	publishAffected    int64
+	lastDescribe       rocketmqDescribeRequest
+	lastFetch          rocketmqFetchRequest
+	lastPublish        rocketmqPublishCommand
+}
+
+func (f *fakeRocketMQRuntime) Close() error { return nil }
+
+func (f *fakeRocketMQRuntime) Ping(ctx context.Context) error { return nil }
+
+func (f *fakeRocketMQRuntime) ListTopics(ctx context.Context, includeSystem bool) ([]rocketmqTopicInfo, error) {
+	result := make([]rocketmqTopicInfo, 0, len(f.listTopicsResult))
+	for _, item := range f.listTopicsResult {
+		if item.System && !includeSystem {
+			continue
+		}
+		result = append(result, item)
+	}
+	return result, nil
+}
+
+func (f *fakeRocketMQRuntime) DescribeTopic(ctx context.Context, request rocketmqDescribeRequest) (rocketmqTopicDescription, error) {
+	f.lastDescribe = request
+	return f.describeResult, nil
+}
+
+func (f *fakeRocketMQRuntime) FetchMessages(ctx context.Context, request rocketmqFetchRequest) ([]rocketmqMessageRecord, error) {
+	f.lastFetch = request
+	items := append([]rocketmqMessageRecord(nil), f.fetchResult...)
+	if request.Offset > 0 {
+		if request.Offset >= len(items) {
+			return []rocketmqMessageRecord{}, nil
+		}
+		items = items[request.Offset:]
+	}
+	if request.Limit > 0 && len(items) > request.Limit {
+		items = items[:request.Limit]
+	}
+	return items, nil
+}
+
+func (f *fakeRocketMQRuntime) Publish(ctx context.Context, command rocketmqPublishCommand) (int64, error) {
+	f.lastPublish = command
+	return f.publishAffected, nil
+}
+
+func TestNormalizeRocketMQConfigParsesURIAndParams(t *testing.T) {
+	config := normalizeRocketMQConfig(connection.ConnectionConfig{
+		URI:              "rocketmq://ak:sk@127.0.0.1:9876,127.0.0.2:9877/orders.events?topology=cluster&groupId=preview&namespace=prod&tag=TagA&pullBatchSize=64&startOffset=latest",
+		ConnectionParams: "producerGroup=writer&sendTimeoutMs=6000",
+	})
+
+	if config.Host != "127.0.0.1" || config.Port != 9876 {
+		t.Fatalf("unexpected rocketmq host/port: %#v", config)
+	}
+	if !reflect.DeepEqual(config.Hosts, []string{"127.0.0.2:9877"}) {
+		t.Fatalf("unexpected rocketmq extra nameservers: %#v", config.Hosts)
+	}
+	if config.User != "ak" || config.Password != "sk" {
+		t.Fatalf("unexpected rocketmq credentials: %#v", config)
+	}
+	if config.Database != "orders.events" || config.Topology != "cluster" {
+		t.Fatalf("unexpected rocketmq topic/topology: %#v", config)
+	}
+
+	params := rocketmqConnectionParams(config)
+	if params.Get("groupId") != "preview" || params.Get("namespace") != "prod" || params.Get("tag") != "TagA" {
+		t.Fatalf("unexpected rocketmq params: %#v", params)
+	}
+	if params.Get("producerGroup") != "writer" || params.Get("sendTimeoutMs") != "6000" {
+		t.Fatalf("unexpected rocketmq producer params: %#v", params)
+	}
+}
+
+func TestRocketMQQueryExecAndColumns(t *testing.T) {
+	fakeRuntime := &fakeRocketMQRuntime{
+		listTopicsResult: []rocketmqTopicInfo{
+			{Name: "orders.events", QueueCount: 2},
+			{Name: "%RETRY%preview", System: true, QueueCount: 1},
+		},
+		describeResult: rocketmqTopicDescription{
+			Name:                  "orders.events",
+			Namespace:             "prod",
+			ConsumerGroup:         "preview",
+			TagExpression:         "*",
+			QueueCount:            2,
+			TotalApproximateCount: 42,
+			Queues: []rocketmqTopicQueueInfo{
+				{BrokerName: "broker-a", QueueID: 0, MinOffset: 0, MaxOffset: 21, ApproximateCount: 21},
+				{BrokerName: "broker-b", QueueID: 1, MinOffset: 0, MaxOffset: 21, ApproximateCount: 21},
+			},
+		},
+		fetchResult: []rocketmqMessageRecord{
+			{
+				Topic:          "orders.events",
+				BrokerName:     "broker-a",
+				QueueID:        0,
+				QueueOffset:    11,
+				MsgID:          "msg-11",
+				OffsetMsgID:    "offset-11",
+				Tags:           "TagA",
+				Keys:           "order-11 tenant-a",
+				Decoded:        map[string]interface{}{"event": "created", "meta": map[string]interface{}{"source": "erp"}},
+				Encoding:       "json",
+				Properties:     map[string]string{"trace": "trace-11"},
+				BornTimestamp:  time.Date(2026, 6, 14, 12, 0, 0, 0, time.UTC),
+				StoreTimestamp: time.Date(2026, 6, 14, 12, 0, 1, 0, time.UTC),
+			},
+			{
+				Topic:          "orders.events",
+				BrokerName:     "broker-b",
+				QueueID:        1,
+				QueueOffset:    12,
+				MsgID:          "msg-12",
+				OffsetMsgID:    "offset-12",
+				Tags:           "TagB",
+				Keys:           "order-12",
+				Decoded:        "plain-text",
+				Encoding:       "text",
+				Properties:     map[string]string{"trace": "trace-12"},
+				BornTimestamp:  time.Date(2026, 6, 14, 12, 0, 2, 0, time.UTC),
+				StoreTimestamp: time.Date(2026, 6, 14, 12, 0, 3, 0, time.UTC),
+			},
+		},
+		publishAffected: 1,
+	}
+
+	originalFactory := newRocketMQRuntime
+	newRocketMQRuntime = func(config connection.ConnectionConfig) (rocketmqRuntime, error) {
+		return fakeRuntime, nil
+	}
+	defer func() {
+		newRocketMQRuntime = originalFactory
+	}()
+
+	client := &RocketMQDB{}
+	if err := client.Connect(connection.ConnectionConfig{
+		Type:             "rocketmq",
+		Host:             "127.0.0.1",
+		Port:             9876,
+		Database:         "orders.events",
+		ConnectionParams: "groupId=preview&namespace=prod&pullBatchSize=48&startOffset=latest",
+	}); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	defer client.Close()
+
+	rows, columns, err := client.Query(`SHOW TOPICS LIMIT 1`)
+	if err != nil {
+		t.Fatalf("SHOW TOPICS failed: %v", err)
+	}
+	if len(rows) != 1 || rows[0]["topic"] != "orders.events" || rows[0]["queue_count"] != 2 {
+		t.Fatalf("unexpected rocketmq topic rows: %#v", rows)
+	}
+	if !containsString(columns, "system_topic") {
+		t.Fatalf("expected system_topic column, got %v", columns)
+	}
+
+	rows, columns, err = client.Query(`DESCRIBE TOPIC "orders.events"`)
+	if err != nil {
+		t.Fatalf("DESCRIBE TOPIC failed: %v", err)
+	}
+	if fakeRuntime.lastDescribe.Topic != "orders.events" || fakeRuntime.lastDescribe.ConsumerGroup != "preview" {
+		t.Fatalf("unexpected describe request: %#v", fakeRuntime.lastDescribe)
+	}
+	if len(rows) != 2 || rows[0]["topic_approximate_count"] != int64(42) {
+		t.Fatalf("unexpected rocketmq describe rows: %#v", rows)
+	}
+	if !containsString(columns, "broker_name") {
+		t.Fatalf("expected broker_name column, got %v", columns)
+	}
+
+	rows, columns, err = client.Query(`SELECT * FROM "orders.events" LIMIT 1 OFFSET 1`)
+	if err != nil {
+		t.Fatalf("SELECT topic failed: %v", err)
+	}
+	if fakeRuntime.lastFetch.Topic != "orders.events" || fakeRuntime.lastFetch.Limit != 1 || fakeRuntime.lastFetch.Offset != 1 || !fakeRuntime.lastFetch.Latest {
+		t.Fatalf("unexpected fetch request: %#v", fakeRuntime.lastFetch)
+	}
+	if len(rows) != 1 || rows[0]["body"] != "plain-text" || rows[0]["properties.trace"] != "trace-12" {
+		t.Fatalf("unexpected rocketmq message rows: %#v", rows)
+	}
+	if !containsString(columns, "body_encoding") || !containsString(columns, "properties.trace") {
+		t.Fatalf("unexpected columns: %v", columns)
+	}
+
+	rows, columns, err = client.Query(`SELECT COUNT(*) FROM "orders.events"`)
+	if err != nil {
+		t.Fatalf("COUNT(*) failed: %v", err)
+	}
+	if len(rows) != 1 || rows[0]["total_approximate_count"] != int64(42) {
+		t.Fatalf("unexpected count rows: %#v", rows)
+	}
+	if !containsString(columns, "queue_count") {
+		t.Fatalf("expected queue_count column, got %v", columns)
+	}
+
+	affected, err := client.Exec(`{"publish":"orders.events","payload":{"id":1},"tag":"TagA","keys":["order-1","tenant-a"],"delayLevel":3,"properties":{"trace":"trace-1"}}`)
+	if err != nil {
+		t.Fatalf("RocketMQ publish failed: %v", err)
+	}
+	if affected != 1 {
+		t.Fatalf("unexpected affected rows: %d", affected)
+	}
+	if fakeRuntime.lastPublish.Topic != "orders.events" || fakeRuntime.lastPublish.Tag != "TagA" || fakeRuntime.lastPublish.DelayLevel != 3 {
+		t.Fatalf("unexpected publish command: %#v", fakeRuntime.lastPublish)
+	}
+	if !reflect.DeepEqual(fakeRuntime.lastPublish.Keys, []string{"order-1", "tenant-a"}) {
+		t.Fatalf("unexpected publish keys: %#v", fakeRuntime.lastPublish.Keys)
+	}
+	if fakeRuntime.lastPublish.Properties["trace"] != "trace-1" {
+		t.Fatalf("unexpected publish properties: %#v", fakeRuntime.lastPublish.Properties)
+	}
+
+	columnDefs, err := client.GetColumns(rocketMQSyntheticDatabase, "orders.events")
+	if err != nil {
+		t.Fatalf("GetColumns failed: %v", err)
+	}
+	names := make([]string, 0, len(columnDefs))
+	for _, col := range columnDefs {
+		names = append(names, col.Name)
+	}
+	joined := strings.Join(names, ",")
+	for _, want := range []string{"topic", "body.meta.source", "properties.trace"} {
+		if !strings.Contains(joined, want) {
+			t.Fatalf("expected rocketmq column %q in %s", want, joined)
+		}
+	}
+
+	databases, err := client.GetDatabases()
+	if err != nil {
+		t.Fatalf("GetDatabases failed: %v", err)
+	}
+	if !reflect.DeepEqual(databases, []string{rocketMQSyntheticDatabase}) {
+		t.Fatalf("unexpected rocketmq database list: %#v", databases)
+	}
+
+	tables, err := client.GetTables(rocketMQSyntheticDatabase)
+	if err != nil {
+		t.Fatalf("GetTables failed: %v", err)
+	}
+	if !reflect.DeepEqual(tables, []string{"orders.events"}) {
+		t.Fatalf("unexpected rocketmq topic list: %#v", tables)
+	}
+}
+
+func TestRocketMQCountRejectsTagFilteredConnections(t *testing.T) {
+	fakeRuntime := &fakeRocketMQRuntime{}
+
+	originalFactory := newRocketMQRuntime
+	newRocketMQRuntime = func(config connection.ConnectionConfig) (rocketmqRuntime, error) {
+		return fakeRuntime, nil
+	}
+	defer func() {
+		newRocketMQRuntime = originalFactory
+	}()
+
+	client := &RocketMQDB{}
+	if err := client.Connect(connection.ConnectionConfig{
+		Type:             "rocketmq",
+		Host:             "127.0.0.1",
+		Port:             9876,
+		Database:         "orders.events",
+		ConnectionParams: "tag=TagA",
+	}); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	defer client.Close()
+
+	if _, _, err := client.Query(`SELECT COUNT(*) FROM "orders.events"`); err == nil || !strings.Contains(err.Error(), "TAG 过滤") {
+		t.Fatalf("expected COUNT(*) to be rejected for tag-filtered RocketMQ, got %v", err)
+	}
+}
+
+func TestRocketMQRecordFromExtUsesPayloadDecoder(t *testing.T) {
+	record := rocketmqRecordFromExt(&rocketmqprimitive.MessageExt{
+		Message: rocketmqprimitive.Message{
+			Topic: "orders.events",
+			Body:  []byte(`{"id":1}`),
+		},
+		MsgId:          "msg-1",
+		OffsetMsgId:    "offset-1",
+		QueueOffset:    2,
+		BornTimestamp:  time.Date(2026, 6, 14, 12, 0, 0, 0, time.UTC).UnixMilli(),
+		StoreTimestamp: time.Date(2026, 6, 14, 12, 0, 1, 0, time.UTC).UnixMilli(),
+	}, "broker-a", 1, 0, 3)
+
+	if record.Encoding != "json" {
+		t.Fatalf("expected json encoding, got %#v", record)
+	}
+	if body, ok := record.Decoded.(map[string]interface{}); !ok || body["id"] == nil {
+		t.Fatalf("unexpected decoded body: %#v", record.Decoded)
+	}
+}