command_mq_topic_list.go 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  1. package shell
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/mq/balancer"
  6. "github.com/seaweedfs/seaweedfs/weed/pb"
  7. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  8. "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
  9. "io"
  10. )
  11. func init() {
  12. Commands = append(Commands, &commandMqTopicList{})
  13. }
  14. type commandMqTopicList struct {
  15. }
  16. func (c *commandMqTopicList) Name() string {
  17. return "mq.topic.list"
  18. }
  19. func (c *commandMqTopicList) Help() string {
  20. return `print out all topics`
  21. }
  22. func (c *commandMqTopicList) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error {
  23. brokerBalancer, err := findBrokerBalancer(commandEnv)
  24. if err != nil {
  25. return err
  26. }
  27. fmt.Fprintf(writer, "current balancer: %s\n", brokerBalancer)
  28. return pb.WithBrokerGrpcClient(false, brokerBalancer, commandEnv.option.GrpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
  29. resp, err := client.ListTopics(context.Background(), &mq_pb.ListTopicsRequest{})
  30. if err != nil {
  31. return err
  32. }
  33. if len(resp.Topics) == 0 {
  34. fmt.Fprintf(writer, "no topics found\n")
  35. return nil
  36. }
  37. for _, topic := range resp.Topics {
  38. fmt.Fprintf(writer, " %+v\n", topic)
  39. }
  40. return nil
  41. })
  42. }
  43. func findBrokerBalancer(commandEnv *CommandEnv) (brokerBalancer string, err error) {
  44. err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
  45. resp, err := client.FindLockOwner(context.Background(), &filer_pb.FindLockOwnerRequest{
  46. Name: balancer.LockBrokerBalancer,
  47. })
  48. if err != nil {
  49. return err
  50. }
  51. brokerBalancer = resp.Owner
  52. return nil
  53. })
  54. return
  55. }