feat(admin): add admin extension

admin/admin_test.go

@@ -0,0 +1,48 @@
+package admin
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/apache/rocketmq-client-go/v2/consumer"
+	"github.com/apache/rocketmq-client-go/v2/internal"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMap(t *testing.T) {
+
+	MQTable := map[string]internal.ProcessQueueInfo{
+		"hahah": {
+			Locked: true,
+		},
+	}
+	data, err := json.Marshal(MQTable)
+	assert.Nil(t, err)
+	t.Log("data info: ", string(data))
+
+	b := map[string]internal.ProcessQueueInfo{}
+	err = json.Unmarshal(data, &b)
+	assert.Nil(t, err)
+	t.Log("b: ", b)
+}
+
+func TestOffset(t *testing.T) {
+
+	MQTable := map[consumer.MessageQueueKey]internal.ProcessQueueInfo{
+		{
+			Topic:      "a",
+			BrokerName: "B-a",
+			QueueId:    1,
+		}: {
+			Locked: true,
+		},
+	}
+	data, err := json.Marshal(MQTable)
+	assert.Nil(t, err)
+	t.Log("data info: ", string(data))
+
+	b := map[consumer.MessageQueueKey]internal.ProcessQueueInfo{}
+	err = json.Unmarshal(data, &b)
+	assert.Nil(t, err)
+	t.Log("b: ", b)
+}

admin/protocol.go

@@ -0,0 +1,38 @@
+/*
+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 (
+	"github.com/apache/rocketmq-client-go/v2/consumer"
+	"github.com/apache/rocketmq-client-go/v2/internal/remote"
+)
+
+type Connection struct {
+	ClientId   string              `json:"clientId"`
+	ClientAddr string              `json:"clientAddr"`
+	Language   remote.LanguageCode `json:"language"`
+	Version    int                 `json:"version"`
+}
+
+// ConsumerConnection consumer connection info fetched from broker, consumer register info by heartbeat
+type ConsumerConnection struct {
+	connectionSet    map[Connection]struct{}
+	consumeType      consumer.ConsumeType
+	messageModel     consumer.MessageModel
+	consumeFromWhere consumer.ConsumeFromWhere
+}

consumer/consumer.go

@@ -277,7 +277,7 @@ func (dc *defaultConsumer) start() error {
 	if dc.model == Clustering {
 		// set retry topic
 		retryTopic := internal.GetRetryTopic(dc.consumerGroup)
-		sub := buildSubscriptionData(retryTopic, MessageSelector{TAG, _SubAll})
+		sub := BuildSubscriptionData(retryTopic, MessageSelector{TAG, _SubAll})
 		dc.subscriptionDataTable.Store(retryTopic, sub)
 	}
 
@@ -336,7 +336,7 @@ func (dc *defaultConsumer) subscriptionAutomatically(topic string) {
 		s := MessageSelector{
 			Expression: _SubAll,
 		}
-		dc.subscriptionDataTable.Store(topic, buildSubscriptionData(topic, s))
+		dc.subscriptionDataTable.Store(topic, BuildSubscriptionData(topic, s))
 	}
 }
 
@@ -979,7 +979,7 @@ func (dc *defaultConsumer) searchOffsetByTimestamp(mq *primitive.MessageQueue, t
 	return strconv.ParseInt(response.ExtFields["offset"], 10, 64)
 }
 
-func buildSubscriptionData(topic string, selector MessageSelector) *internal.SubscriptionData {
+func BuildSubscriptionData(topic string, selector MessageSelector) *internal.SubscriptionData {
 	subData := &internal.SubscriptionData{
 		Topic:     topic,
 		SubString: selector.Expression,
@@ -1014,7 +1014,7 @@ func buildSubscriptionData(topic string, selector MessageSelector) *internal.Sub
 	return subData
 }
 
-func buildSysFlag(commitOffset, suspend, subscription, classFilter bool) int32 {
+func BuildSysFlag(commitOffset, suspend, subscription, classFilter bool) int32 {
 	var flag int32 = 0
 	if commitOffset {
 		flag |= 0x1 << 0

consumer/pull_consumer.go

@@ -119,7 +119,7 @@ func (c *defaultPullConsumer) Pull(ctx context.Context, topic string, selector M
 		return nil, fmt.Errorf("prepard to pull topic: %s, but no queue is founded", topic)
 	}
 
-	data := buildSubscriptionData(mq.Topic, selector)
+	data := BuildSubscriptionData(mq.Topic, selector)
 	result, err := c.pull(context.Background(), mq, data, c.nextOffsetOf(mq), numbers)
 
 	if err != nil {
@@ -197,7 +197,7 @@ func (c *defaultPullConsumer) pull(ctx context.Context, mq *primitive.MessageQue
 
 	c.subscriptionAutomatically(mq.Topic)
 
-	sysFlag := buildSysFlag(false, true, true, false)
+	sysFlag := BuildSysFlag(false, true, true, false)
 
 	pullResp, err := c.pullInner(ctx, mq, data, offset, numbers, sysFlag, 0)
 	if err != nil {
@@ -226,7 +226,7 @@ func (c *defaultPullConsumer) PullFrom(ctx context.Context, queue *primitive.Mes
 	}
 
 	selector := MessageSelector{}
-	data := buildSubscriptionData(queue.Topic, selector)
+	data := BuildSubscriptionData(queue.Topic, selector)
 
 	return c.pull(ctx, queue, data, offset, numbers)
 }

consumer/push_consumer.go

@@ -228,7 +228,7 @@ func (pc *pushConsumer) Subscribe(topic string, selector MessageSelector,
 	if pc.option.Namespace != "" {
 		topic = pc.option.Namespace + "%" + topic
 	}
-	data := buildSubscriptionData(topic, selector)
+	data := BuildSubscriptionData(topic, selector)
 	pc.subscriptionDataTable.Store(topic, data)
 	pc.subscribedTopic[topic] = ""
 
@@ -612,7 +612,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 			subExpression = sd.SubString
 		}
 
-		sysFlag := buildSysFlag(commitOffsetEnable, true, subExpression != "", classFilter)
+		sysFlag := BuildSysFlag(commitOffsetEnable, true, subExpression != "", classFilter)
 
 		pullRequest := &internal.PullMessageRequestHeader{
 			ConsumerGroup:        pc.consumerGroup,