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.

109 lines
3.2 KiB

5 years ago
5 years ago
5 years ago
5 years ago
3 years ago
3 years ago
5 years ago
5 years ago
5 years ago
3 years ago
3 years ago
3 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
2 years ago
5 years ago
2 years ago
5 years ago
3 years ago
5 years ago
5 years ago
  1. package broker
  2. import (
  3. "github.com/seaweedfs/seaweedfs/weed/cluster"
  4. "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
  5. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  6. "google.golang.org/grpc"
  7. "time"
  8. "github.com/seaweedfs/seaweedfs/weed/pb"
  9. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  10. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  11. )
  12. type MessageQueueBrokerOption struct {
  13. Masters map[string]pb.ServerAddress
  14. FilerGroup string
  15. DataCenter string
  16. Rack string
  17. DefaultReplication string
  18. MaxMB int
  19. Ip string
  20. Port int
  21. Cipher bool
  22. }
  23. type MessageQueueBroker struct {
  24. mq_pb.UnimplementedSeaweedMessagingServer
  25. option *MessageQueueBrokerOption
  26. grpcDialOption grpc.DialOption
  27. MasterClient *wdclient.MasterClient
  28. filers map[pb.ServerAddress]struct{}
  29. currentFiler pb.ServerAddress
  30. }
  31. func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (mqBroker *MessageQueueBroker, err error) {
  32. mqBroker = &MessageQueueBroker{
  33. option: option,
  34. grpcDialOption: grpcDialOption,
  35. MasterClient: wdclient.NewMasterClient(grpcDialOption, option.FilerGroup, cluster.BrokerType, pb.NewServerAddress(option.Ip, option.Port, 0), option.DataCenter, option.Rack, option.Masters),
  36. filers: make(map[pb.ServerAddress]struct{}),
  37. }
  38. mqBroker.MasterClient.OnPeerUpdate = mqBroker.OnBrokerUpdate
  39. go mqBroker.MasterClient.KeepConnectedToMaster()
  40. existingNodes := cluster.ListExistingPeerUpdates(mqBroker.MasterClient.GetMaster(), grpcDialOption, option.FilerGroup, cluster.FilerType)
  41. for _, newNode := range existingNodes {
  42. mqBroker.OnBrokerUpdate(newNode, time.Now())
  43. }
  44. return mqBroker, nil
  45. }
  46. func (broker *MessageQueueBroker) OnBrokerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
  47. if update.NodeType != cluster.FilerType {
  48. return
  49. }
  50. address := pb.ServerAddress(update.Address)
  51. if update.IsAdd {
  52. broker.filers[address] = struct{}{}
  53. if broker.currentFiler == "" {
  54. broker.currentFiler = address
  55. }
  56. } else {
  57. delete(broker.filers, address)
  58. if broker.currentFiler == address {
  59. for filer, _ := range broker.filers {
  60. broker.currentFiler = filer
  61. break
  62. }
  63. }
  64. }
  65. }
  66. func (broker *MessageQueueBroker) GetFiler() pb.ServerAddress {
  67. return broker.currentFiler
  68. }
  69. func (broker *MessageQueueBroker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
  70. return pb.WithFilerClient(streamingMode, broker.GetFiler(), broker.grpcDialOption, fn)
  71. }
  72. func (broker *MessageQueueBroker) AdjustedUrl(location *filer_pb.Location) string {
  73. return location.Url
  74. }
  75. func (broker *MessageQueueBroker) withMasterClient(streamingMode bool, master pb.ServerAddress, fn func(client master_pb.SeaweedClient) error) error {
  76. return pb.WithMasterClient(streamingMode, master, broker.grpcDialOption, func(client master_pb.SeaweedClient) error {
  77. return fn(client)
  78. })
  79. }
  80. func (broker *MessageQueueBroker) withBrokerClient(streamingMode bool, server pb.ServerAddress, fn func(client mq_pb.SeaweedMessagingClient) error) error {
  81. return pb.WithBrokerClient(streamingMode, server, broker.grpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
  82. return fn(client)
  83. })
  84. }