From 94b8c42b2cba40557ab28cd613ab8c19a93fca23 Mon Sep 17 00:00:00 2001 From: chrislu Date: Sun, 10 Jul 2022 12:11:37 -0700 Subject: [PATCH] clean up --- weed/cluster/cluster.go | 22 ++-- weed/command/mq_broker.go | 27 +++-- weed/command/server.go | 2 + weed/mq/broker/brokder_grpc_admin.go | 10 ++ weed/mq/broker/consistent_distribution.go | 38 ------ .../mq/broker/consistent_distribution_test.go | 32 ----- weed/pb/mq.proto | 9 +- weed/pb/mq_pb/mq.pb.go | 112 +++++++----------- weed/pb/mq_pb/mq_grpc.pb.go | 28 ++--- 9 files changed, 97 insertions(+), 183 deletions(-) create mode 100644 weed/mq/broker/brokder_grpc_admin.go delete mode 100644 weed/mq/broker/consistent_distribution.go delete mode 100644 weed/mq/broker/consistent_distribution_test.go diff --git a/weed/cluster/cluster.go b/weed/cluster/cluster.go index 0d9886911..445b89a36 100644 --- a/weed/cluster/cluster.go +++ b/weed/cluster/cluster.go @@ -17,7 +17,7 @@ const ( type FilerGroup string type Filers struct { - filers map[pb.ServerAddress]*ClusterNode + members map[pb.ServerAddress]*ClusterNode leaders *Leaders } type Leaders struct { @@ -60,7 +60,7 @@ func (cluster *Cluster) getFilers(filerGroup FilerGroup, createIfNotFound bool) filers, found := cluster.filerGroup2filers[filerGroup] if !found && createIfNotFound { filers = &Filers{ - filers: make(map[pb.ServerAddress]*ClusterNode), + members: make(map[pb.ServerAddress]*ClusterNode), leaders: &Leaders{}, } cluster.filerGroup2filers[filerGroup] = filers @@ -75,11 +75,11 @@ func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCente cluster.filersLock.Lock() defer cluster.filersLock.Unlock() filers := cluster.getFilers(filerGroup, true) - if existingNode, found := filers.filers[address]; found { + if existingNode, found := filers.members[address]; found { existingNode.counter++ return nil } - filers.filers[address] = &ClusterNode{ + filers.members[address] = &ClusterNode{ Address: address, Version: version, counter: 1, @@ -87,7 +87,7 @@ func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCente DataCenter: dataCenter, Rack: rack, } - return cluster.ensureFilerLeaders(filers, true, filerGroup, nodeType, address) + return ensureFilerLeaders(filers, true, filerGroup, nodeType, address) case BrokerType: cluster.brokersLock.Lock() defer cluster.brokersLock.Unlock() @@ -151,13 +151,13 @@ func (cluster *Cluster) RemoveClusterNode(ns string, nodeType string, dataCenter if filers == nil { return nil } - if existingNode, found := filers.filers[address]; !found { + if existingNode, found := filers.members[address]; !found { return nil } else { existingNode.counter-- if existingNode.counter <= 0 { - delete(filers.filers, address) - return cluster.ensureFilerLeaders(filers, false, filerGroup, nodeType, address) + delete(filers.members, address) + return ensureFilerLeaders(filers, false, filerGroup, nodeType, address) } } case BrokerType: @@ -215,7 +215,7 @@ func (cluster *Cluster) ListClusterNode(filerGroup FilerGroup, nodeType string) if filers == nil { return } - for _, node := range filers.filers { + for _, node := range filers.members { nodes = append(nodes, node) } case BrokerType: @@ -241,7 +241,7 @@ func (cluster *Cluster) IsOneLeader(filerGroup FilerGroup, address pb.ServerAddr return filers.leaders.isOneLeader(address) } -func (cluster *Cluster) ensureFilerLeaders(filers *Filers, isAdd bool, filerGroup FilerGroup, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) { +func ensureFilerLeaders(filers *Filers, isAdd bool, filerGroup FilerGroup, nodeType string, address pb.ServerAddress) (result []*master_pb.KeepConnectedResponse) { if isAdd { if filers.leaders.addLeaderIfVacant(address) { // has added the address as one leader @@ -282,7 +282,7 @@ func (cluster *Cluster) ensureFilerLeaders(filers *Filers, isAdd bool, filerGrou var shortestDuration int64 = math.MaxInt64 now := time.Now() var candidateAddress pb.ServerAddress - for _, node := range filers.filers { + for _, node := range filers.members { if filers.leaders.isOneLeader(node.Address) { continue } diff --git a/weed/command/mq_broker.go b/weed/command/mq_broker.go index 65f49e673..cce77bf8e 100644 --- a/weed/command/mq_broker.go +++ b/weed/command/mq_broker.go @@ -23,20 +23,21 @@ var ( ) type MessageQueueBrokerOptions struct { - masters *string - filerGroup *string - filer *string - ip *string - port *int - dataCenter *string - rack *string - cpuprofile *string - memprofile *string + masters map[string]pb.ServerAddress + mastersString *string + filerGroup *string + filer *string + ip *string + port *int + dataCenter *string + rack *string + cpuprofile *string + memprofile *string } func init() { cmdMqBroker.Run = runMqBroker // break init cycle - mqBrokerStandaloneOptions.masters = cmdMqBroker.Flag.String("master", "localhost:9333", "comma-separated master servers") + mqBrokerStandaloneOptions.mastersString = cmdMqBroker.Flag.String("master", "localhost:9333", "comma-separated master servers") mqBrokerStandaloneOptions.filer = cmdMqBroker.Flag.String("filer", "localhost:8888", "filer server address") mqBrokerStandaloneOptions.filerGroup = cmdMqBroker.Flag.String("filerGroup", "", "share metadata with other filers in the same filerGroup") mqBrokerStandaloneOptions.ip = cmdMqBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address") @@ -48,7 +49,7 @@ func init() { } var cmdMqBroker = &Command{ - UsageLine: "mq.broker [-port=17777] [-filer=]", + UsageLine: "mq.broker [-port=17777] [-master=]", Short: "start a message queue broker", Long: `start a message queue broker @@ -62,6 +63,8 @@ func runMqBroker(cmd *Command, args []string) bool { util.LoadConfiguration("security", false) + mqBrokerStandaloneOptions.masters = pb.ServerAddresses(*mqBrokerStandaloneOptions.mastersString).ToAddressMap() + return mqBrokerStandaloneOptions.startQueueServer() } @@ -94,7 +97,7 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool { } qs, err := broker.NewMessageBroker(&broker.MessageQueueBrokerOption{ - Masters: pb.ServerAddresses(*mqBrokerOpt.masters).ToAddressMap(), + Masters: mqBrokerOpt.masters, FilerGroup: *mqBrokerOpt.filerGroup, DataCenter: *mqBrokerOpt.dataCenter, Rack: *mqBrokerOpt.rack, diff --git a/weed/command/server.go b/weed/command/server.go index 9e1cea07d..04b07ed51 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -209,6 +209,8 @@ func runServer(cmd *Command, args []string) bool { serverOptions.v.dataCenter = serverDataCenter serverOptions.v.rack = serverRack mqBrokerOptions.ip = serverIp + mqBrokerOptions.masters = filerOptions.masters + mqBrokerOptions.filerGroup = filerOptions.filerGroup // serverOptions.v.pulseSeconds = pulseSeconds // masterOptions.pulseSeconds = pulseSeconds diff --git a/weed/mq/broker/brokder_grpc_admin.go b/weed/mq/broker/brokder_grpc_admin.go new file mode 100644 index 000000000..e8914cbed --- /dev/null +++ b/weed/mq/broker/brokder_grpc_admin.go @@ -0,0 +1,10 @@ +package broker + +import ( + "context" + "github.com/chrislusf/seaweedfs/weed/pb/mq_pb" +) + +func (broker *MessageQueueBroker) FindBrokerLeader(c context.Context, request *mq_pb.FindBrokerLeaderRequest) (*mq_pb.FindBrokerLeaderResponse, error) { + panic("implement me") +} diff --git a/weed/mq/broker/consistent_distribution.go b/weed/mq/broker/consistent_distribution.go deleted file mode 100644 index 465a2a8f2..000000000 --- a/weed/mq/broker/consistent_distribution.go +++ /dev/null @@ -1,38 +0,0 @@ -package broker - -import ( - "github.com/buraksezer/consistent" - "github.com/cespare/xxhash" -) - -type Member string - -func (m Member) String() string { - return string(m) -} - -type hasher struct{} - -func (h hasher) Sum64(data []byte) uint64 { - return xxhash.Sum64(data) -} - -func PickMember(members []string, key []byte) string { - cfg := consistent.Config{ - PartitionCount: 9791, - ReplicationFactor: 2, - Load: 1.25, - Hasher: hasher{}, - } - - cmembers := []consistent.Member{} - for _, m := range members { - cmembers = append(cmembers, Member(m)) - } - - c := consistent.New(cmembers, cfg) - - m := c.LocateKey(key) - - return m.String() -} diff --git a/weed/mq/broker/consistent_distribution_test.go b/weed/mq/broker/consistent_distribution_test.go deleted file mode 100644 index f58fe4e0e..000000000 --- a/weed/mq/broker/consistent_distribution_test.go +++ /dev/null @@ -1,32 +0,0 @@ -package broker - -import ( - "fmt" - "testing" -) - -func TestPickMember(t *testing.T) { - - servers := []string{ - "s1:port", - "s2:port", - "s3:port", - "s5:port", - "s4:port", - } - - total := 1000 - - distribution := make(map[string]int) - for i := 0; i < total; i++ { - tp := fmt.Sprintf("tp:%2d", i) - m := PickMember(servers, []byte(tp)) - // println(tp, "=>", m) - distribution[m]++ - } - - for member, count := range distribution { - fmt.Printf("member: %s, key count: %d load=%.2f\n", member, count, float64(count*100)/float64(total/len(servers))) - } - -} diff --git a/weed/pb/mq.proto b/weed/pb/mq.proto index 44291c173..47bd1775e 100644 --- a/weed/pb/mq.proto +++ b/weed/pb/mq.proto @@ -10,19 +10,16 @@ option java_outer_classname = "MessagQueueProto"; service SeaweedMessaging { - rpc FindBroker (FindBrokerRequest) returns (FindBrokerResponse) { + rpc FindBrokerLeader (FindBrokerLeader) returns (FindBrokerLeaderResponse) { } } ////////////////////////////////////////////////// -message FindBrokerRequest { - string namespace = 1; - string topic = 2; - int32 parition = 3; +message FindBrokerLeaderRequest { } -message FindBrokerResponse { +message FindBrokerLeaderResponse { string broker = 1; } diff --git a/weed/pb/mq_pb/mq.pb.go b/weed/pb/mq_pb/mq.pb.go index dac27c959..a5d63c8d9 100644 --- a/weed/pb/mq_pb/mq.pb.go +++ b/weed/pb/mq_pb/mq.pb.go @@ -20,18 +20,14 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -type FindBrokerRequest struct { +type FindBrokerLeaderRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - - Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` - Topic string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"` - Parition int32 `protobuf:"varint,3,opt,name=parition,proto3" json:"parition,omitempty"` } -func (x *FindBrokerRequest) Reset() { - *x = FindBrokerRequest{} +func (x *FindBrokerLeaderRequest) Reset() { + *x = FindBrokerLeaderRequest{} if protoimpl.UnsafeEnabled { mi := &file_mq_proto_msgTypes[0] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -39,13 +35,13 @@ func (x *FindBrokerRequest) Reset() { } } -func (x *FindBrokerRequest) String() string { +func (x *FindBrokerLeaderRequest) String() string { return protoimpl.X.MessageStringOf(x) } -func (*FindBrokerRequest) ProtoMessage() {} +func (*FindBrokerLeaderRequest) ProtoMessage() {} -func (x *FindBrokerRequest) ProtoReflect() protoreflect.Message { +func (x *FindBrokerLeaderRequest) ProtoReflect() protoreflect.Message { mi := &file_mq_proto_msgTypes[0] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -57,33 +53,12 @@ func (x *FindBrokerRequest) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use FindBrokerRequest.ProtoReflect.Descriptor instead. -func (*FindBrokerRequest) Descriptor() ([]byte, []int) { +// Deprecated: Use FindBrokerLeaderRequest.ProtoReflect.Descriptor instead. +func (*FindBrokerLeaderRequest) Descriptor() ([]byte, []int) { return file_mq_proto_rawDescGZIP(), []int{0} } -func (x *FindBrokerRequest) GetNamespace() string { - if x != nil { - return x.Namespace - } - return "" -} - -func (x *FindBrokerRequest) GetTopic() string { - if x != nil { - return x.Topic - } - return "" -} - -func (x *FindBrokerRequest) GetParition() int32 { - if x != nil { - return x.Parition - } - return 0 -} - -type FindBrokerResponse struct { +type FindBrokerLeaderResponse struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields @@ -91,8 +66,8 @@ type FindBrokerResponse struct { Broker string `protobuf:"bytes,1,opt,name=broker,proto3" json:"broker,omitempty"` } -func (x *FindBrokerResponse) Reset() { - *x = FindBrokerResponse{} +func (x *FindBrokerLeaderResponse) Reset() { + *x = FindBrokerLeaderResponse{} if protoimpl.UnsafeEnabled { mi := &file_mq_proto_msgTypes[1] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -100,13 +75,13 @@ func (x *FindBrokerResponse) Reset() { } } -func (x *FindBrokerResponse) String() string { +func (x *FindBrokerLeaderResponse) String() string { return protoimpl.X.MessageStringOf(x) } -func (*FindBrokerResponse) ProtoMessage() {} +func (*FindBrokerLeaderResponse) ProtoMessage() {} -func (x *FindBrokerResponse) ProtoReflect() protoreflect.Message { +func (x *FindBrokerLeaderResponse) ProtoReflect() protoreflect.Message { mi := &file_mq_proto_msgTypes[1] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -118,12 +93,12 @@ func (x *FindBrokerResponse) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use FindBrokerResponse.ProtoReflect.Descriptor instead. -func (*FindBrokerResponse) Descriptor() ([]byte, []int) { +// Deprecated: Use FindBrokerLeaderResponse.ProtoReflect.Descriptor instead. +func (*FindBrokerLeaderResponse) Descriptor() ([]byte, []int) { return file_mq_proto_rawDescGZIP(), []int{1} } -func (x *FindBrokerResponse) GetBroker() string { +func (x *FindBrokerLeaderResponse) GetBroker() string { if x != nil { return x.Broker } @@ -134,28 +109,25 @@ var File_mq_proto protoreflect.FileDescriptor var file_mq_proto_rawDesc = []byte{ 0x0a, 0x08, 0x6d, 0x71, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x6d, 0x65, 0x73, 0x73, - 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x22, 0x63, 0x0a, 0x11, 0x46, 0x69, 0x6e, 0x64, - 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, - 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, - 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, - 0x63, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x72, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, - 0x01, 0x28, 0x05, 0x52, 0x08, 0x70, 0x61, 0x72, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x2c, 0x0a, - 0x12, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, - 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x18, 0x01, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x32, 0x65, 0x0a, 0x10, 0x53, - 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x12, - 0x51, 0x0a, 0x0a, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x1f, 0x2e, - 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x46, 0x69, 0x6e, - 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, - 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x46, 0x69, - 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, - 0x22, 0x00, 0x42, 0x4e, 0x0a, 0x0c, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, - 0x6d, 0x71, 0x42, 0x10, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x51, 0x75, 0x65, 0x75, 0x65, 0x50, - 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, - 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61, 0x77, 0x65, - 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x6d, 0x71, 0x5f, - 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x22, 0x19, 0x0a, 0x17, 0x46, 0x69, 0x6e, 0x64, + 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x4c, 0x65, 0x61, 0x64, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x22, 0x32, 0x0a, 0x18, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, + 0x72, 0x4c, 0x65, 0x61, 0x64, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, + 0x16, 0x0a, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x32, 0x77, 0x0a, 0x10, 0x53, 0x65, 0x61, 0x77, 0x65, + 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x12, 0x63, 0x0a, 0x10, 0x46, + 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x4c, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, + 0x25, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x46, + 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x4c, 0x65, 0x61, 0x64, 0x65, 0x72, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, + 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, + 0x4c, 0x65, 0x61, 0x64, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, + 0x42, 0x4e, 0x0a, 0x0c, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, 0x6d, 0x71, + 0x42, 0x10, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x51, 0x75, 0x65, 0x75, 0x65, 0x50, 0x72, 0x6f, + 0x74, 0x6f, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, + 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, + 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x6d, 0x71, 0x5f, 0x70, 0x62, + 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -172,12 +144,12 @@ func file_mq_proto_rawDescGZIP() []byte { var file_mq_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_mq_proto_goTypes = []interface{}{ - (*FindBrokerRequest)(nil), // 0: messaging_pb.FindBrokerRequest - (*FindBrokerResponse)(nil), // 1: messaging_pb.FindBrokerResponse + (*FindBrokerLeaderRequest)(nil), // 0: messaging_pb.FindBrokerLeaderRequest + (*FindBrokerLeaderResponse)(nil), // 1: messaging_pb.FindBrokerLeaderResponse } var file_mq_proto_depIdxs = []int32{ - 0, // 0: messaging_pb.SeaweedMessaging.FindBroker:input_type -> messaging_pb.FindBrokerRequest - 1, // 1: messaging_pb.SeaweedMessaging.FindBroker:output_type -> messaging_pb.FindBrokerResponse + 0, // 0: messaging_pb.SeaweedMessaging.FindBrokerLeader:input_type -> messaging_pb.FindBrokerLeaderRequest + 1, // 1: messaging_pb.SeaweedMessaging.FindBrokerLeader:output_type -> messaging_pb.FindBrokerLeaderResponse 1, // [1:2] is the sub-list for method output_type 0, // [0:1] is the sub-list for method input_type 0, // [0:0] is the sub-list for extension type_name @@ -192,7 +164,7 @@ func file_mq_proto_init() { } if !protoimpl.UnsafeEnabled { file_mq_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*FindBrokerRequest); i { + switch v := v.(*FindBrokerLeaderRequest); i { case 0: return &v.state case 1: @@ -204,7 +176,7 @@ func file_mq_proto_init() { } } file_mq_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*FindBrokerResponse); i { + switch v := v.(*FindBrokerLeaderResponse); i { case 0: return &v.state case 1: diff --git a/weed/pb/mq_pb/mq_grpc.pb.go b/weed/pb/mq_pb/mq_grpc.pb.go index 60258e23d..5851632cc 100644 --- a/weed/pb/mq_pb/mq_grpc.pb.go +++ b/weed/pb/mq_pb/mq_grpc.pb.go @@ -18,7 +18,7 @@ const _ = grpc.SupportPackageIsVersion7 // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type SeaweedMessagingClient interface { - FindBroker(ctx context.Context, in *FindBrokerRequest, opts ...grpc.CallOption) (*FindBrokerResponse, error) + FindBrokerLeader(ctx context.Context, in *FindBrokerLeaderRequest, opts ...grpc.CallOption) (*FindBrokerLeaderResponse, error) } type seaweedMessagingClient struct { @@ -29,9 +29,9 @@ func NewSeaweedMessagingClient(cc grpc.ClientConnInterface) SeaweedMessagingClie return &seaweedMessagingClient{cc} } -func (c *seaweedMessagingClient) FindBroker(ctx context.Context, in *FindBrokerRequest, opts ...grpc.CallOption) (*FindBrokerResponse, error) { - out := new(FindBrokerResponse) - err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/FindBroker", in, out, opts...) +func (c *seaweedMessagingClient) FindBrokerLeader(ctx context.Context, in *FindBrokerLeaderRequest, opts ...grpc.CallOption) (*FindBrokerLeaderResponse, error) { + out := new(FindBrokerLeaderResponse) + err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/FindBrokerLeader", in, out, opts...) if err != nil { return nil, err } @@ -42,7 +42,7 @@ func (c *seaweedMessagingClient) FindBroker(ctx context.Context, in *FindBrokerR // All implementations must embed UnimplementedSeaweedMessagingServer // for forward compatibility type SeaweedMessagingServer interface { - FindBroker(context.Context, *FindBrokerRequest) (*FindBrokerResponse, error) + FindBrokerLeader(context.Context, *FindBrokerLeaderRequest) (*FindBrokerLeaderResponse, error) mustEmbedUnimplementedSeaweedMessagingServer() } @@ -50,8 +50,8 @@ type SeaweedMessagingServer interface { type UnimplementedSeaweedMessagingServer struct { } -func (UnimplementedSeaweedMessagingServer) FindBroker(context.Context, *FindBrokerRequest) (*FindBrokerResponse, error) { - return nil, status.Errorf(codes.Unimplemented, "method FindBroker not implemented") +func (UnimplementedSeaweedMessagingServer) FindBrokerLeader(context.Context, *FindBrokerLeaderRequest) (*FindBrokerLeaderResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method FindBrokerLeader not implemented") } func (UnimplementedSeaweedMessagingServer) mustEmbedUnimplementedSeaweedMessagingServer() {} @@ -66,20 +66,20 @@ func RegisterSeaweedMessagingServer(s grpc.ServiceRegistrar, srv SeaweedMessagin s.RegisterService(&SeaweedMessaging_ServiceDesc, srv) } -func _SeaweedMessaging_FindBroker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { - in := new(FindBrokerRequest) +func _SeaweedMessaging_FindBrokerLeader_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(FindBrokerLeaderRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { - return srv.(SeaweedMessagingServer).FindBroker(ctx, in) + return srv.(SeaweedMessagingServer).FindBrokerLeader(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/messaging_pb.SeaweedMessaging/FindBroker", + FullMethod: "/messaging_pb.SeaweedMessaging/FindBrokerLeader", } handler := func(ctx context.Context, req interface{}) (interface{}, error) { - return srv.(SeaweedMessagingServer).FindBroker(ctx, req.(*FindBrokerRequest)) + return srv.(SeaweedMessagingServer).FindBrokerLeader(ctx, req.(*FindBrokerLeaderRequest)) } return interceptor(ctx, in, info, handler) } @@ -92,8 +92,8 @@ var SeaweedMessaging_ServiceDesc = grpc.ServiceDesc{ HandlerType: (*SeaweedMessagingServer)(nil), Methods: []grpc.MethodDesc{ { - MethodName: "FindBroker", - Handler: _SeaweedMessaging_FindBroker_Handler, + MethodName: "FindBrokerLeader", + Handler: _SeaweedMessaging_FindBrokerLeader_Handler, }, }, Streams: []grpc.StreamDesc{},