add some admin tool apis

admin/admin.go

@@ -30,11 +30,21 @@ import (
 )
 
 type Admin interface {
+	// create one topic
 	CreateTopic(ctx context.Context, opts ...OptionCreate) error
+
+	// delete one topic
 	DeleteTopic(ctx context.Context, opts ...OptionDelete) error
-	//TODO
-	//TopicList(ctx context.Context, mq *primitive.MessageQueue) (*remote.RemotingCommand, error)
-	//GetBrokerClusterInfo(ctx context.Context) (*remote.RemotingCommand, error)
+
+	// fetch all topics
+	FetchAllTopicList(ctx context.Context) (*internal.TopicList, error)
+
+	// get cluster info
+	GetBrokerClusterInfo(ctx context.Context) (*remote.RemotingCommand, error)
+
+	// fetch topic queues
+	FetchPublishMessageQueues(ctx context.Context, topic string) ([]*primitive.MessageQueue, error)
+
 	Close() error
 }
 
@@ -70,7 +80,7 @@ type admin struct {
 }
 
 // NewAdmin initialize admin
-func NewAdmin(opts ...AdminOption) (Admin, error) {
+func NewAdmin(opts ...AdminOption) (*admin, error) {
 	defaultOpts := defaultAdminOptions()
 	for _, opt := range opts {
 		opt(defaultOpts)
@@ -202,6 +212,20 @@ func (a *admin) DeleteTopic(ctx context.Context, opts ...OptionDelete) error {
 	return nil
 }
 
+// fetch topic queues
+func (a *admin) FetchPublishMessageQueues(ctx context.Context, topic string) ([]*primitive.MessageQueue, error) {
+	return a.cli.GetNameSrv().FetchPublishMessageQueues(topic)
+}
+
+// fetch all topic names
+func (a *admin) FetchAllTopicList(ctx context.Context) (*internal.TopicList, error) {
+	return a.cli.GetNameSrv().FetchAllTopicList(ctx)
+}
+
+// get cluster info from namesrv
+func (a *admin) GetBrokerClusterInfo(ctx context.Context) (*internal.ClusterInfo, error) {
+	return a.cli.GetBrokerClusterInfo(ctx)
+}
 func (a *admin) Close() error {
 	a.closeOnce.Do(func() {
 		a.cli.Shutdown()

admin/admin_test.go

@@ -0,0 +1,59 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package admin
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+var nameSrvAddr = []string{"127.0.0.1:9876"}
+
+func TestFetchAllTopicList(t *testing.T) {
+	adminTool, err := NewAdmin(WithResolver(primitive.NewPassthroughResolver(nameSrvAddr)))
+	assert.True(t, err == nil)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+	topicNames, error := adminTool.FetchAllTopicList(context.Background())
+	assert.True(t, error == nil)
+	assert.True(t, len(topicNames.TopicNameList) > 0)
+
+	prettyJson, _ := json.MarshalIndent(topicNames, "", " ")
+	fmt.Println(string(prettyJson))
+}
+
+func TestGetBrokerClusterInfo(t *testing.T) {
+	adminTool, err := NewAdmin(WithResolver(primitive.NewPassthroughResolver(nameSrvAddr)))
+	assert.True(t, err == nil)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+	cluster, err := adminTool.GetBrokerClusterInfo(context.Background())
+	assert.True(t, err == nil)
+	assert.True(t, cluster != nil)
+
+	prettyJson, _ := json.MarshalIndent(cluster, "", " ")
+	fmt.Println(string(prettyJson))
+}

internal/client.go

@@ -161,6 +161,8 @@ type RMQClient interface {
 	UpdatePublishInfo(topic string, data *TopicRouteData, changed bool)
 
 	GetNameSrv() Namesrvs
+
+	GetBrokerClusterInfo(ctx context.Context) (*ClusterInfo, error)
 }
 
 var _ RMQClient = new(rmqClient)
@@ -924,6 +926,24 @@ func (c *rmqClient) consumeMessageDirectly(msg *primitive.MessageExt, group stri
 	return res
 }
 
+func (c *rmqClient) GetBrokerClusterInfo(ctx context.Context) (*ClusterInfo, error) {
+	request := remote.NewRemotingCommand(ReqGetBrokerClusterInfo, nil, nil)
+	c.SendHeartbeatToAllBrokerWithLock()
+
+	responseCommand, _ := c.remoteClient.InvokeSync(ctx, c.GetNameSrv().AddrList()[0], request)
+
+	switch responseCommand.Code {
+	case ResSuccess:
+		if responseCommand.Body != nil {
+			var cluster, err = ParseClusterInfo(string(responseCommand.Body))
+			return cluster, err
+		}
+	default:
+		rlog.Warning("no any topic list", nil)
+	}
+	return nil, errors.New("no any response from broker")
+
+}
 func routeData2SubscribeInfo(topic string, data *TopicRouteData) []*primitive.MessageQueue {
 	list := make([]*primitive.MessageQueue, 0)
 	for idx := range data.QueueDataList {

internal/cluster.go

@@ -0,0 +1,119 @@
+package internal
+
+import (
+	"encoding/json"
+	"github.com/apache/rocketmq-client-go/v2/rlog"
+	"github.com/pkg/errors"
+	"github.com/tidwall/gjson"
+	"gopkg.in/yaml.v2"
+	"strings"
+)
+
+type ClusterInfo struct {
+	BrokerAddrTable  map[string]BrokerData `json:"brokerAddrTable"`
+	ClusterAddrTable map[string][]string   `json:"clusterAddrTable"`
+}
+
+type ClusterInfoMap struct {
+	BrokerAddrTable  map[string]string `json:"brokerAddrTable"`
+	ClusterAddrTable map[string]string `json:"clusterAddrTable"`
+}
+
+func ParseClusterInfo(jsonString string) (cluster *ClusterInfo, errResult error) {
+	var brokerAddrTableResult = gjson.Get(jsonString, "brokerAddrTable")
+	if !brokerAddrTableResult.Exists() {
+		return nil, errors.New("json key for brokerAddrTable not exist")
+	}
+	var clusterAddrTableResult = gjson.Get(jsonString, "clusterAddrTable")
+	if !clusterAddrTableResult.Exists() {
+		return nil, errors.New("json key for clusterAddrTable not exist")
+	}
+
+	var mapAndArray, err = ParseClusterAddrTable(clusterAddrTableResult.String())
+	if err != nil {
+		return nil, err
+	}
+
+	if len(mapAndArray) <= 0 {
+		return nil, nil
+	}
+
+	var clusterInfo = &ClusterInfo{}
+	clusterInfo.ClusterAddrTable, errResult = ParseClusterAddrTable(clusterAddrTableResult.String())
+	if errResult != nil {
+		return nil, errResult
+	}
+	clusterInfo.BrokerAddrTable, errResult = ParseBrokerAddrTable(brokerAddrTableResult.String(), clusterInfo.ClusterAddrTable)
+	if errResult != nil {
+		return nil, errResult
+	}
+	return clusterInfo, nil
+}
+
+// parse broker addr table
+// input like : {"broker-a":{"brokerAddrs":{0:"127.0.0.1:10911"},"brokerName":"broker-a","cluster":"DefaultCluster"}}
+// result.key = broker name, result.value=BrokerData
+func ParseBrokerAddrTable(jsonString string, clusterAndBrokerNamesMap map[string][]string) (map[string]BrokerData, error) {
+	var brokerMap = make(map[string]BrokerData)
+	for _, brokerNames := range clusterAndBrokerNamesMap {
+		for _, brokerName := range brokerNames {
+			var brokerInfo = gjson.Get(jsonString, brokerName).String()
+			var brokerAddrsString = gjson.Get(brokerInfo, "brokerAddrs").String() // {0:"x.x.x.x:10911"}
+			var brokerName1 = gjson.Get(brokerInfo, "brokerName").String()
+			var cluster = gjson.Get(brokerInfo, "cluster").String()
+
+			bd := BrokerData{
+				Cluster:         cluster,
+				BrokerName:      brokerName1,
+				BrokerAddresses: ParseBrokerAddrs(brokerAddrsString),
+			}
+
+			brokerMap[brokerName1] = bd
+		}
+	}
+
+	return brokerMap, nil
+}
+
+// input like : "{0:"127.0.0.1:10911"}" to map directly,can't parse, so get one by one
+// result.key = broker id , result.value = broker address and port
+func ParseBrokerAddrs1(jsonString string) map[int64]string {
+	var resultMap = make(map[int64]string)
+	var broker0Addr = gjson.Get(jsonString, "0")
+	if broker0Addr.Exists() {
+		resultMap[0] = broker0Addr.String()
+	}
+	var broker1Addr = gjson.Get(jsonString, "1")
+	if broker1Addr.Exists() {
+		resultMap[1] = broker1Addr.String()
+	}
+	return resultMap
+}
+
+// input like : "{0:"127.0.0.1:10911"}" to map directly,can't parse, so get one by one
+// result.key = broker id , result.value = broker address and port
+func ParseBrokerAddrs(jsonString string) map[int64]string {
+	// for yaml parse, do replace
+	jsonString = strings.ReplaceAll(jsonString, "0:", "0: ")
+	jsonString = strings.ReplaceAll(jsonString, "1:", "1: ")
+
+	var yamlmap map[int64]string
+	err := yaml.Unmarshal([]byte(jsonString), &yamlmap)
+	if err != nil {
+		rlog.Error("parse addr error "+jsonString, nil)
+		return nil
+	}
+
+	return yamlmap
+}
+
+// input like : {"DefaultCluster":["broker-a"]}
+// result.key = cluster name, result.value = broker names
+func ParseClusterAddrTable(jsonString string) (map[string][]string, error) {
+	var result = make(map[string][]string)
+	err := json.Unmarshal([]byte(jsonString), &result)
+	if err != nil {
+		return nil, err
+	}
+	return result, nil
+}

internal/cluster_test.go

@@ -0,0 +1,30 @@
+package internal
+
+import (
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"gopkg.in/yaml.v2"
+	"testing"
+)
+
+func TestParseBrokerAddrs(t *testing.T) {
+	str := `{0: "127.0.0.1:1111",1: "127.0.0.1:1112"}`
+	blob := []byte(str)
+	var yamlmap map[int]string
+	yaml.Unmarshal(blob, &yamlmap)
+	for k, v := range yamlmap {
+		fmt.Println(fmt.Sprintf("k=%d, v=%s", k, v))
+	}
+
+}
+
+func TestJsonToMap(t *testing.T) {
+	var jsonString = []byte(`{"brokerAddrTable":{"broker-a":{"brokerAddrs":{0:"127.0.0.1:10911"},"brokerName":"broker-a","cluster":"DefaultCluster"}},"clusterAddrTable":{"DefaultCluster":["broker-a"]}}`)
+	cluster, err := ParseClusterInfo(string(jsonString))
+
+	var cl = &ClusterInfo{}
+	yaml.Unmarshal(jsonString, cl)
+
+	assert.True(t, err == nil)
+	fmt.Println(cluster)
+}

internal/namesrv.go

@@ -18,8 +18,11 @@ limitations under the License.
 package internal
 
 import (
+	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/rlog"
 	"regexp"
 	"strings"
 	"sync"
@@ -62,6 +65,8 @@ type Namesrvs interface {
 
 	FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error)
 
+	FetchAllTopicList(ctx context.Context) (*TopicList, error)
+
 	AddrList() []string
 }
 
@@ -153,6 +158,7 @@ func (s *namesrvs) Size() int {
 func (s *namesrvs) String() string {
 	return strings.Join(s.srvs, ";")
 }
+
 func (s *namesrvs) SetCredentials(credentials primitive.Credentials) {
 	s.nameSrvClient.RegisterInterceptor(remote.ACLInterceptor(credentials))
 }
@@ -179,3 +185,28 @@ func (s *namesrvs) UpdateNameServerAddress() {
 
 	s.srvs = srvs
 }
+
+func (s *namesrvs) FetchAllTopicList(ctx context.Context) (lists *TopicList, err error) {
+	request := remote.NewRemotingCommand(ReqGetAllTopicListFromNameServer, nil, nil)
+	var responseCommand *remote.RemotingCommand
+
+	responseCommand, err = s.nameSrvClient.InvokeSync(ctx, s.getNameServerAddress(), request) // brokerId =0 is master
+	if err != nil {
+		rlog.Error(fmt.Sprintf("FetchAllTopicList error from %s, %+v", s.getNameServerAddress(), err), nil)
+	}
+	var topicNames = &TopicList{}
+	switch responseCommand.Code {
+	case ResSuccess:
+		if responseCommand.Body == nil {
+			return nil, errors.New("FetchAllTopicList return body is nil")
+		}
+		err := json.Unmarshal(responseCommand.Body, topicNames)
+		if err != nil {
+			return nil, err
+		}
+		return topicNames, nil
+	default:
+		rlog.Warning("no any topic list", nil)
+	}
+	return nil, errors.New("empty topic list")
+}

internal/topic_list.go

@@ -0,0 +1,8 @@
+package internal
+
+// topic list struct
+type TopicList struct {
+	// topic names
+	TopicNameList []string `json:"topicList"`
+
+}