create topic and report topicorigin/sub

1 files changed, 16 insertions, 13 deletions

diff --git a/weed/shell/command_mq_topic_list.go b/weed/shell/command_mq_topic_list.go
index 5d18d8755..5fb913a8a 100644
--- a/weed/shell/command_mq_topic_list.go
+++ b/weed/shell/command_mq_topic_list.go
@@ -4,7 +4,9 @@ import (
 	"context"
 	"fmt"
 	"github.com/seaweedfs/seaweedfs/weed/mq/balancer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"io"
 )
 
@@ -29,21 +31,22 @@ func (c *commandMqTopicList) Do(args []string, commandEnv *CommandEnv, writer io
 	if err != nil {
 		return err
 	}
-
-	//pb.WithBrokerGrpcClient(false, brokerBalancer, commandEnv.option.GrpcDialOption, func(client pb.SeaweedMessagingClient) error {
-	//	resp, err := client.ListTopics(context.Background(), &pb.ListTopicsRequest{})
-	//	if err != nil {
-	//		return err
-	//	}
-	//	for _, topic := range resp.Topics {
-	//		fmt.Fprintf(writer, "%s\n", topic)
-	//	}
-	//	return nil
-	//})
-
 	fmt.Fprintf(writer, "current balancer: %s\n", brokerBalancer)
 
-	return nil
+	return pb.WithBrokerGrpcClient(false, brokerBalancer, commandEnv.option.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
+		resp, err := client.ListTopics(context.Background(), &mq_pb.ListTopicsRequest{})
+		if err != nil {
+			return err
+		}
+		if len(resp.Topics) == 0 {
+			fmt.Fprintf(writer, "no topics found\n")
+			return nil
+		}
+		for _, topic := range resp.Topics {
+			fmt.Fprintf(writer, "  %+v\n", topic)
+		}
+		return nil
+	})
 }
 
 func findBrokerBalancer(commandEnv *CommandEnv) (brokerBalancer string, err error) {