add CreateTopic API

This commit is contained in:
chrislu
2023-09-24 14:22:11 -07:00
parent 0f8168c0c9
commit 0361c321b4
9 changed files with 714 additions and 400 deletions

View File

@@ -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
}