You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

85 lines
2.9 KiB

1 year ago
1 year ago
1 year ago
1 year ago
1 year ago
  1. package broker
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
  6. "google.golang.org/grpc/codes"
  7. "google.golang.org/grpc/status"
  8. )
  9. // FindTopicBrokers returns the brokers that are serving the topic
  10. //
  11. // 1. lock the topic
  12. //
  13. // 2. find the topic partitions on the filer
  14. // 2.1 if the topic is not found, return error
  15. // 2.2 if the request is_for_publish, create the topic
  16. // 2.2.1 if the request is_for_subscribe, return error not found
  17. // 2.2.2 if the request is_for_publish, create the topic
  18. // 2.2 if the topic is found, return the brokers
  19. //
  20. // 3. unlock the topic
  21. func (broker *MessageQueueBroker) LookupTopicBrokers(ctx context.Context, request *mq_pb.LookupTopicBrokersRequest) (resp *mq_pb.LookupTopicBrokersResponse, err error) {
  22. if broker.currentBalancer == "" {
  23. return nil, status.Errorf(codes.Unavailable, "no balancer")
  24. }
  25. if !broker.lockAsBalancer.IsLocked() {
  26. proxyErr := broker.withBrokerClient(false, broker.currentBalancer, func(client mq_pb.SeaweedMessagingClient) error {
  27. resp, err = client.LookupTopicBrokers(ctx, request)
  28. return nil
  29. })
  30. if proxyErr != nil {
  31. return nil, proxyErr
  32. }
  33. return resp, err
  34. }
  35. ret := &mq_pb.LookupTopicBrokersResponse{}
  36. ret.Topic = request.Topic
  37. ret.BrokerPartitionAssignments, err = broker.Balancer.LookupOrAllocateTopicPartitions(ret.Topic, request.IsForPublish, 6)
  38. return ret, err
  39. }
  40. // CheckTopicPartitionsStatus check the topic partitions on the broker
  41. func (broker *MessageQueueBroker) CheckTopicPartitionsStatus(c context.Context, request *mq_pb.CheckTopicPartitionsStatusRequest) (*mq_pb.CheckTopicPartitionsStatusResponse, error) {
  42. ret := &mq_pb.CheckTopicPartitionsStatusResponse{}
  43. return ret, nil
  44. }
  45. func (broker *MessageQueueBroker) ListTopics(ctx context.Context, request *mq_pb.ListTopicsRequest) (resp *mq_pb.ListTopicsResponse, err error) {
  46. if broker.currentBalancer == "" {
  47. return nil, status.Errorf(codes.Unavailable, "no balancer")
  48. }
  49. if !broker.lockAsBalancer.IsLocked() {
  50. proxyErr := broker.withBrokerClient(false, broker.currentBalancer, func(client mq_pb.SeaweedMessagingClient) error {
  51. resp, err = client.ListTopics(ctx, request)
  52. return nil
  53. })
  54. if proxyErr != nil {
  55. return nil, proxyErr
  56. }
  57. return resp, err
  58. }
  59. ret := &mq_pb.ListTopicsResponse{}
  60. knownTopics := make(map[string]struct{})
  61. for brokerStatsItem := range broker.Balancer.Brokers.IterBuffered() {
  62. _, brokerStats := brokerStatsItem.Key, brokerStatsItem.Val
  63. for topicPartitionStatsItem := range brokerStats.Stats.IterBuffered() {
  64. topicPartitionStat := topicPartitionStatsItem.Val
  65. topic := &mq_pb.Topic{
  66. Namespace: topicPartitionStat.TopicPartition.Namespace,
  67. Name: topicPartitionStat.TopicPartition.Name,
  68. }
  69. topicKey := fmt.Sprintf("%s/%s", topic.Namespace, topic.Name)
  70. if _, found := knownTopics[topicKey]; found {
  71. continue
  72. }
  73. knownTopics[topicKey] = struct{}{}
  74. ret.Topics = append(ret.Topics, topic)
  75. }
  76. }
  77. return ret, nil
  78. }