Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

clean up

pull/3379/head
chrislu 2 years ago
parent
8060fdcac5
commit
94b8c42b2c
9 changed files with 97 additions and 183 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 22
      weed/cluster/cluster.go
  2. 27
      weed/command/mq_broker.go
  3. 2
      weed/command/server.go
  4. 10
      weed/mq/broker/brokder_grpc_admin.go
  5. 38
      weed/mq/broker/consistent_distribution.go
  6. 32
      weed/mq/broker/consistent_distribution_test.go
  7. 9
      weed/pb/mq.proto
  8. 112
      weed/pb/mq_pb/mq.pb.go
  9. 28
      weed/pb/mq_pb/mq_grpc.pb.go

22
weed/cluster/cluster.go
View File

@ -17,7 +17,7 @@ const (
type FilerGroup string type FilerGroup string
type Filers struct { type Filers struct {
filers map[pb.ServerAddress]*ClusterNode
members map[pb.ServerAddress]*ClusterNode
leaders *Leaders leaders *Leaders
} }
type Leaders struct { type Leaders struct {
@ -60,7 +60,7 @@ func (cluster *Cluster) getFilers(filerGroup FilerGroup, createIfNotFound bool)
filers, found := cluster.filerGroup2filers[filerGroup] filers, found := cluster.filerGroup2filers[filerGroup]
if !found && createIfNotFound { if !found && createIfNotFound {
filers = &Filers{ filers = &Filers{
filers: make(map[pb.ServerAddress]*ClusterNode),
members: make(map[pb.ServerAddress]*ClusterNode),
leaders: &Leaders{}, leaders: &Leaders{},
} }
cluster.filerGroup2filers[filerGroup] = filers cluster.filerGroup2filers[filerGroup] = filers
@ -75,11 +75,11 @@ func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCente
cluster.filersLock.Lock() cluster.filersLock.Lock()
defer cluster.filersLock.Unlock() defer cluster.filersLock.Unlock()
filers := cluster.getFilers(filerGroup, true) filers := cluster.getFilers(filerGroup, true)
if existingNode, found := filers.filers[address]; found {
if existingNode, found := filers.members[address]; found {
existingNode.counter++ existingNode.counter++
return nil return nil
} }
filers.filers[address] = &ClusterNode{
filers.members[address] = &ClusterNode{
Address: address, Address: address,
Version: version, Version: version,
counter: 1, counter: 1,
@ -87,7 +87,7 @@ func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCente
DataCenter: dataCenter, DataCenter: dataCenter,
Rack: rack, Rack: rack,
} }
return cluster.ensureFilerLeaders(filers, true, filerGroup, nodeType, address)
return ensureFilerLeaders(filers, true, filerGroup, nodeType, address)
case BrokerType: case BrokerType:
cluster.brokersLock.Lock() cluster.brokersLock.Lock()
defer cluster.brokersLock.Unlock() defer cluster.brokersLock.Unlock()
@ -151,13 +151,13 @@ func (cluster *Cluster) RemoveClusterNode(ns string, nodeType string, dataCenter
if filers == nil { if filers == nil {
return nil return nil
} }
if existingNode, found := filers.filers[address]; !found {
if existingNode, found := filers.members[address]; !found {
return nil return nil
} else { } else {
existingNode.counter-- existingNode.counter--
if existingNode.counter <= 0 { 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: case BrokerType:
@ -215,7 +215,7 @@ func (cluster *Cluster) ListClusterNode(filerGroup FilerGroup, nodeType string)
if filers == nil { if filers == nil {
return return
} }
for _, node := range filers.filers {
for _, node := range filers.members {
nodes = append(nodes, node) nodes = append(nodes, node)
} }
case BrokerType: case BrokerType:
@ -241,7 +241,7 @@ func (cluster *Cluster) IsOneLeader(filerGroup FilerGroup, address pb.ServerAddr
return filers.leaders.isOneLeader(address) 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 isAdd {
if filers.leaders.addLeaderIfVacant(address) { if filers.leaders.addLeaderIfVacant(address) {
// has added the address as one leader // 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 var shortestDuration int64 = math.MaxInt64
now := time.Now() now := time.Now()
var candidateAddress pb.ServerAddress var candidateAddress pb.ServerAddress
for _, node := range filers.filers {
for _, node := range filers.members {
if filers.leaders.isOneLeader(node.Address) { if filers.leaders.isOneLeader(node.Address) {
continue continue
} }

27
weed/command/mq_broker.go
View File

@ -23,20 +23,21 @@ var (
) )
type MessageQueueBrokerOptions struct { 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() { func init() {
cmdMqBroker.Run = runMqBroker // break init cycle 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.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.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") mqBrokerStandaloneOptions.ip = cmdMqBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address")
@ -48,7 +49,7 @@ func init() {
} }
var cmdMqBroker = &Command{ var cmdMqBroker = &Command{
UsageLine: "mq.broker [-port=17777] [-filer=<ip:port>]",
UsageLine: "mq.broker [-port=17777] [-master=<ip:port>]",
Short: "start a message queue broker", Short: "start a message queue broker",
Long: `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) util.LoadConfiguration("security", false)
mqBrokerStandaloneOptions.masters = pb.ServerAddresses(*mqBrokerStandaloneOptions.mastersString).ToAddressMap()
return mqBrokerStandaloneOptions.startQueueServer() return mqBrokerStandaloneOptions.startQueueServer()
} }
@ -94,7 +97,7 @@ func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
} }
qs, err := broker.NewMessageBroker(&broker.MessageQueueBrokerOption{ qs, err := broker.NewMessageBroker(&broker.MessageQueueBrokerOption{
Masters: pb.ServerAddresses(*mqBrokerOpt.masters).ToAddressMap(),
Masters: mqBrokerOpt.masters,
FilerGroup: *mqBrokerOpt.filerGroup, FilerGroup: *mqBrokerOpt.filerGroup,
DataCenter: *mqBrokerOpt.dataCenter, DataCenter: *mqBrokerOpt.dataCenter,
Rack: *mqBrokerOpt.rack, Rack: *mqBrokerOpt.rack,

2
weed/command/server.go
View File

@ -209,6 +209,8 @@ func runServer(cmd *Command, args []string) bool {
serverOptions.v.dataCenter = serverDataCenter serverOptions.v.dataCenter = serverDataCenter
serverOptions.v.rack = serverRack serverOptions.v.rack = serverRack
mqBrokerOptions.ip = serverIp mqBrokerOptions.ip = serverIp
mqBrokerOptions.masters = filerOptions.masters
mqBrokerOptions.filerGroup = filerOptions.filerGroup
// serverOptions.v.pulseSeconds = pulseSeconds // serverOptions.v.pulseSeconds = pulseSeconds
// masterOptions.pulseSeconds = pulseSeconds // masterOptions.pulseSeconds = pulseSeconds

10
weed/mq/broker/brokder_grpc_admin.go
View File

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

38
weed/mq/broker/consistent_distribution.go
View File

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

32
weed/mq/broker/consistent_distribution_test.go
View File

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

9
weed/pb/mq.proto
View File

@ -10,19 +10,16 @@ option java_outer_classname = "MessagQueueProto";
service SeaweedMessaging { 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; string broker = 1;
} }

112
weed/pb/mq_pb/mq.pb.go
View File

@ -20,18 +20,14 @@ const (
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
) )
type FindBrokerRequest struct {
type FindBrokerLeaderRequest struct {
state protoimpl.MessageState state protoimpl.MessageState
sizeCache protoimpl.SizeCache sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields 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 { if protoimpl.UnsafeEnabled {
mi := &file_mq_proto_msgTypes[0] mi := &file_mq_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 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) 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] mi := &file_mq_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil { if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@ -57,33 +53,12 @@ func (x *FindBrokerRequest) ProtoReflect() protoreflect.Message {
return mi.MessageOf(x) 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} 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 state protoimpl.MessageState
sizeCache protoimpl.SizeCache sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields unknownFields protoimpl.UnknownFields
@ -91,8 +66,8 @@ type FindBrokerResponse struct {
Broker string `protobuf:"bytes,1,opt,name=broker,proto3" json:"broker,omitempty"` 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 { if protoimpl.UnsafeEnabled {
mi := &file_mq_proto_msgTypes[1] mi := &file_mq_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 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) 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] mi := &file_mq_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil { if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@ -118,12 +93,12 @@ func (x *FindBrokerResponse) ProtoReflect() protoreflect.Message {
return mi.MessageOf(x) 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} return file_mq_proto_rawDescGZIP(), []int{1}
} }
func (x *FindBrokerResponse) GetBroker() string {
func (x *FindBrokerLeaderResponse) GetBroker() string {
if x != nil { if x != nil {
return x.Broker return x.Broker
} }
@ -134,28 +109,25 @@ var File_mq_proto protoreflect.FileDescriptor
var file_mq_proto_rawDesc = []byte{ var file_mq_proto_rawDesc = []byte{
0x0a, 0x08, 0x6d, 0x71, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x6d, 0x65, 0x73, 0x73, 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 ( var (
@ -172,12 +144,12 @@ func file_mq_proto_rawDescGZIP() []byte {
var file_mq_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_mq_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_mq_proto_goTypes = []interface{}{ 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{ 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 1, // [1:2] is the sub-list for method output_type
0, // [0:1] is the sub-list for method input_type 0, // [0:1] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name 0, // [0:0] is the sub-list for extension type_name
@ -192,7 +164,7 @@ func file_mq_proto_init() {
} }
if !protoimpl.UnsafeEnabled { if !protoimpl.UnsafeEnabled {
file_mq_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { file_mq_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*FindBrokerRequest); i {
switch v := v.(*FindBrokerLeaderRequest); i {
case 0: case 0:
return &v.state return &v.state
case 1: case 1:
@ -204,7 +176,7 @@ func file_mq_proto_init() {
} }
} }
file_mq_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { file_mq_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*FindBrokerResponse); i {
switch v := v.(*FindBrokerLeaderResponse); i {
case 0: case 0:
return &v.state return &v.state
case 1: case 1:

28
weed/pb/mq_pb/mq_grpc.pb.go
View File

@ -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. // 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 { 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 { type seaweedMessagingClient struct {
@ -29,9 +29,9 @@ func NewSeaweedMessagingClient(cc grpc.ClientConnInterface) SeaweedMessagingClie
return &seaweedMessagingClient{cc} 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 { if err != nil {
return nil, err return nil, err
} }
@ -42,7 +42,7 @@ func (c *seaweedMessagingClient) FindBroker(ctx context.Context, in *FindBrokerR
// All implementations must embed UnimplementedSeaweedMessagingServer // All implementations must embed UnimplementedSeaweedMessagingServer
// for forward compatibility // for forward compatibility
type SeaweedMessagingServer interface { type SeaweedMessagingServer interface {
FindBroker(context.Context, *FindBrokerRequest) (*FindBrokerResponse, error)
FindBrokerLeader(context.Context, *FindBrokerLeaderRequest) (*FindBrokerLeaderResponse, error)
mustEmbedUnimplementedSeaweedMessagingServer() mustEmbedUnimplementedSeaweedMessagingServer()
} }
@ -50,8 +50,8 @@ type SeaweedMessagingServer interface {
type UnimplementedSeaweedMessagingServer struct { 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() {} func (UnimplementedSeaweedMessagingServer) mustEmbedUnimplementedSeaweedMessagingServer() {}
@ -66,20 +66,20 @@ func RegisterSeaweedMessagingServer(s grpc.ServiceRegistrar, srv SeaweedMessagin
s.RegisterService(&SeaweedMessaging_ServiceDesc, srv) 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 { if err := dec(in); err != nil {
return nil, err return nil, err
} }
if interceptor == nil { if interceptor == nil {
return srv.(SeaweedMessagingServer).FindBroker(ctx, in)
return srv.(SeaweedMessagingServer).FindBrokerLeader(ctx, in)
} }
info := &grpc.UnaryServerInfo{ info := &grpc.UnaryServerInfo{
Server: srv, Server: srv,
FullMethod: "/messaging_pb.SeaweedMessaging/FindBroker",
FullMethod: "/messaging_pb.SeaweedMessaging/FindBrokerLeader",
} }
handler := func(ctx context.Context, req interface{}) (interface{}, error) { 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) return interceptor(ctx, in, info, handler)
} }
@ -92,8 +92,8 @@ var SeaweedMessaging_ServiceDesc = grpc.ServiceDesc{
HandlerType: (*SeaweedMessagingServer)(nil), HandlerType: (*SeaweedMessagingServer)(nil),
Methods: []grpc.MethodDesc{ Methods: []grpc.MethodDesc{
{ {
MethodName: "FindBroker",
Handler: _SeaweedMessaging_FindBroker_Handler,
MethodName: "FindBrokerLeader",
Handler: _SeaweedMessaging_FindBrokerLeader_Handler,
}, },
}, },
Streams: []grpc.StreamDesc{}, Streams: []grpc.StreamDesc{},

Write Preview
Loading…
Cancel
Save