add CreateTopic API

1 files changed, 35 insertions, 2 deletions

diff --git a/weed/shell/command_mq_topic_list.go b/weed/shell/command_mq_topic_list.go
index a069b0614..5d18d8755 100644
--- a/weed/shell/command_mq_topic_list.go
+++ b/weed/shell/command_mq_topic_list.go
@@ -1,7 +1,10 @@
 package shell
 
 import (
+	"context"
 	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/mq/balancer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"io"
 )
 
@@ -20,9 +23,39 @@ func (c *commandMqTopicList) Help() string {
 	return `print out all topics`
 }
 
-func (c *commandMqTopicList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+func (c *commandMqTopicList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
 
-	fmt.Fprintf(writer, "%s\n", commandEnv.option.Directory)
+	brokerBalancer, err := findBrokerBalancer(commandEnv)
+	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
 }
+
+func findBrokerBalancer(commandEnv *CommandEnv) (brokerBalancer string, err error) {
+	err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := client.FindLockOwner(context.Background(), &filer_pb.FindLockOwnerRequest{
+			Name: balancer.LockBrokerBalancer,
+		})
+		if err != nil {
+			return err
+		}
+		brokerBalancer = resp.Owner
+		return nil
+	})
+	return
+}