diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 6d890861d..3b3b78bbb 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -48,6 +48,8 @@ service SeaweedFiler { rpc SubscribeMetadata (SubscribeMetadataRequest) returns (stream SubscribeMetadataResponse) { } + rpc KeepConnected (stream KeepConnectedRequest) returns (stream KeepConnectedResponse) { + } } ////////////////////////////////////////////////// @@ -261,3 +263,10 @@ message LogEntry { int32 partition_key_hash = 2; bytes data = 3; } + +message KeepConnectedRequest { + string name = 1; + uint32 grpc_port = 2; +} +message KeepConnectedResponse { +} diff --git a/weed/command/msg_broker.go b/weed/command/msg_broker.go index 984497c2e..bab1083ab 100644 --- a/weed/command/msg_broker.go +++ b/weed/command/msg_broker.go @@ -6,9 +6,10 @@ import ( "strconv" "time" - "github.com/chrislusf/seaweedfs/weed/util/grace" "google.golang.org/grpc/reflection" + "github.com/chrislusf/seaweedfs/weed/util/grace" + "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/messaging/broker" "github.com/chrislusf/seaweedfs/weed/pb" @@ -24,6 +25,7 @@ var ( type QueueOptions struct { filer *string + ip *string port *int cpuprofile *string memprofile *string @@ -32,7 +34,8 @@ type QueueOptions struct { func init() { cmdMsgBroker.Run = runMsgBroker // break init cycle messageBrokerStandaloneOptions.filer = cmdMsgBroker.Flag.String("filer", "localhost:8888", "filer server address") - messageBrokerStandaloneOptions.port = cmdMsgBroker.Flag.Int("port", 17777, "queue server gRPC listen port") + messageBrokerStandaloneOptions.ip = cmdMsgBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address") + messageBrokerStandaloneOptions.port = cmdMsgBroker.Flag.Int("port", 17777, "broker gRPC listen port") messageBrokerStandaloneOptions.cpuprofile = cmdMsgBroker.Flag.String("cpuprofile", "", "cpu profile output file") messageBrokerStandaloneOptions.memprofile = cmdMsgBroker.Flag.String("memprofile", "", "memory profile output file") } @@ -91,6 +94,7 @@ func (msgBrokerOpt *QueueOptions) startQueueServer() bool { Filers: []string{*msgBrokerOpt.filer}, DefaultReplication: "", MaxMB: 0, + Ip: *msgBrokerOpt.ip, Port: *msgBrokerOpt.port, Cipher: cipher, }, grpcDialOption) diff --git a/weed/messaging/broker/broker_grpc_server_discovery.go b/weed/messaging/broker/broker_grpc_server_discovery.go new file mode 100644 index 000000000..4b7f357fa --- /dev/null +++ b/weed/messaging/broker/broker_grpc_server_discovery.go @@ -0,0 +1,87 @@ +package broker + +import ( + "context" + "time" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/pb/master_pb" + "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb" +) + +/* +Topic discovery: + +When pub or sub connects, it ask for the whole broker list, and run consistent hashing to find the broker. + +The broker will check peers whether it is already hosted by some other broker, if that broker is alive and acknowledged alive, redirect to it. +Otherwise, just host the topic. + +So, if the pub or sub connects around the same time, they would connect to the same broker. Everyone is happy. +If one of the pub or sub connects very late, and the system topo changed quite a bit with new servers added or old servers died, checking peers will help. + +*/ + +func (broker *MessageBroker) FindBroker(c context.Context, request *messaging_pb.FindBrokerRequest) (*messaging_pb.FindBrokerResponse, error) { + + panic("implement me") +} + + + +func (broker *MessageBroker) checkPeers() { + + // contact a filer about masters + var masters []string + found := false + for !found { + for _, filer := range broker.option.Filers { + err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error { + resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + if err != nil { + return err + } + masters = append(masters, resp.Masters...) + return nil + }) + if err == nil { + found = true + break + } + glog.V(0).Infof("failed to read masters from %+v: %v", broker.option.Filers, err) + time.Sleep(time.Second) + } + } + glog.V(0).Infof("received master list: %s", masters) + + // contact each masters for filers + var filers []string + found = false + for !found { + for _, master := range masters { + err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error { + resp, err := client.ListMasterClients(context.Background(), &master_pb.ListMasterClientsRequest{ + ClientType: "filer", + }) + if err != nil { + return err + } + + filers = append(filers, resp.GrpcAddresses...) + + return nil + }) + if err == nil { + found = true + break + } + glog.V(0).Infof("failed to list filers: %v", err) + time.Sleep(time.Second) + } + } + glog.V(0).Infof("received filer list: %s", filers) + + broker.option.Filers = filers + +} diff --git a/weed/messaging/broker/broker_grpc_server_subscribe.go b/weed/messaging/broker/broker_grpc_server_subscribe.go index 379063eed..472a5007b 100644 --- a/weed/messaging/broker/broker_grpc_server_subscribe.go +++ b/weed/messaging/broker/broker_grpc_server_subscribe.go @@ -38,7 +38,6 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs } if err = stream.Send(&messaging_pb.BrokerMessage{ - Redirect: nil, }); err != nil { return err } diff --git a/weed/messaging/broker/broker_server.go b/weed/messaging/broker/broker_server.go index 29c227274..9cad27214 100644 --- a/weed/messaging/broker/broker_server.go +++ b/weed/messaging/broker/broker_server.go @@ -16,6 +16,7 @@ type MessageBrokerOption struct { Filers []string DefaultReplication string MaxMB int + Ip string Port int Cipher bool } @@ -37,73 +38,44 @@ func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOptio messageBroker.checkPeers() - // go messageBroker.loopForEver() + go messageBroker.keepConnectedToOneFiler() return messageBroker, nil } -func (broker *MessageBroker) loopForEver() { +func (broker *MessageBroker) keepConnectedToOneFiler() { for { - broker.checkPeers() - time.Sleep(3 * time.Second) - } - -} - -func (broker *MessageBroker) checkPeers() { - - // contact a filer about masters - var masters []string - found := false - for !found { for _, filer := range broker.option.Filers { - err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error { + stream, err := client.KeepConnected(context.Background()) if err != nil { + glog.V(0).Infof("%s:%d failed to keep connected to %s: %v", broker.option.Ip, broker.option.Port, filer, err) return err } - masters = append(masters, resp.Masters...) - return nil - }) - if err == nil { - found = true - break - } - glog.V(0).Infof("failed to read masters from %+v: %v", broker.option.Filers, err) - time.Sleep(time.Second) - } - } - glog.V(0).Infof("received master list: %s", masters) - - // contact each masters for filers - var filers []string - found = false - for !found { - for _, master := range masters { - err := broker.withMasterClient(master, func(client master_pb.SeaweedClient) error { - resp, err := client.ListMasterClients(context.Background(), &master_pb.ListMasterClientsRequest{ - ClientType: "filer", - }) - if err != nil { - return err + glog.V(0).Infof("conntected with filer: %v", filer) + for { + if err := stream.Send(&filer_pb.KeepConnectedRequest{ + Name: broker.option.Ip, + GrpcPort: uint32(broker.option.Port), + }); err != nil { + glog.V(0).Infof("%s:%d failed to sendto %s: %v", broker.option.Ip, broker.option.Port, filer, err) + return err + } + // println("send heartbeat") + if _, err := stream.Recv(); err != nil { + glog.V(0).Infof("%s:%d failed to receive from %s: %v", broker.option.Ip, broker.option.Port, filer, err) + return err + } + // println("received reply") + time.Sleep(11*time.Second) + // println("woke up") } - - filers = append(filers, resp.GrpcAddresses...) - return nil }) - if err == nil { - found = true - break - } - glog.V(0).Infof("failed to list filers: %v", err) - time.Sleep(time.Second) + time.Sleep(3*time.Second) } } - glog.V(0).Infof("received filer list: %s", filers) - - broker.option.Filers = filers } diff --git a/weed/messaging/client/client.go b/weed/messaging/client/client.go deleted file mode 100644 index 4a674a9fc..000000000 --- a/weed/messaging/client/client.go +++ /dev/null @@ -1,30 +0,0 @@ -package client - -import ( - "context" - - "google.golang.org/grpc" - - "github.com/chrislusf/seaweedfs/weed/pb" - "github.com/chrislusf/seaweedfs/weed/security" - "github.com/chrislusf/seaweedfs/weed/util" -) - -type MessagingClient struct { - bootstrapBrokers []string - grpcConnection *grpc.ClientConn -} - -func NewMessagingClient(bootstrapBrokers []string) (*MessagingClient, error) { - grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.msg_client") - - grpcConnection, err := pb.GrpcDial(context.Background(), "localhost:17777", grpcDialOption) - if err != nil { - return nil, err - } - - return &MessagingClient{ - bootstrapBrokers: bootstrapBrokers, - grpcConnection: grpcConnection, - }, nil -} diff --git a/weed/messaging/msgclient/client.go b/weed/messaging/msgclient/client.go new file mode 100644 index 000000000..f4e11232e --- /dev/null +++ b/weed/messaging/msgclient/client.go @@ -0,0 +1,56 @@ +package msgclient + +import ( + "context" + "fmt" + "log" + + "google.golang.org/grpc" + + "github.com/chrislusf/seaweedfs/weed/messaging/broker" + "github.com/chrislusf/seaweedfs/weed/pb" + "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb" + "github.com/chrislusf/seaweedfs/weed/security" + "github.com/chrislusf/seaweedfs/weed/util" +) + +type MessagingClient struct { + bootstrapBrokers []string + grpcConnections map[broker.TopicPartition]*grpc.ClientConn + grpcDialOption grpc.DialOption +} + +func NewMessagingClient(bootstrapBrokers ...string) *MessagingClient { + return &MessagingClient{ + bootstrapBrokers: bootstrapBrokers, + grpcConnections: make(map[broker.TopicPartition]*grpc.ClientConn), + grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.msg_client"), + } +} + + +func (mc *MessagingClient) findBroker(tp broker.TopicPartition) (*grpc.ClientConn, error) { + + for _, broker := range mc.bootstrapBrokers { + grpcConnection, err := pb.GrpcDial(context.Background(), broker, mc.grpcDialOption) + if err != nil { + log.Printf("dial broker %s: %v", broker, err) + continue + } + defer grpcConnection.Close() + + resp, err := messaging_pb.NewSeaweedMessagingClient(grpcConnection).FindBroker(context.Background(), + &messaging_pb.FindBrokerRequest{ + Namespace: tp.Namespace, + Topic: tp.Topic, + Parition: tp.Partition, + }) + if err != nil { + return nil, err + } + + targetBroker := resp.Broker + return pb.GrpcDial(context.Background(), targetBroker, mc.grpcDialOption) + } + return nil, fmt.Errorf("no broker found for %+v", tp) +} diff --git a/weed/messaging/msgclient/pub_sub_chan.go b/weed/messaging/msgclient/pub_sub_chan.go new file mode 100644 index 000000000..d39e4c658 --- /dev/null +++ b/weed/messaging/msgclient/pub_sub_chan.go @@ -0,0 +1,96 @@ +package msgclient + +import ( + "io" + "log" + "time" + + "github.com/chrislusf/seaweedfs/weed/messaging/broker" + "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb" +) + +type PubChannel struct { + client messaging_pb.SeaweedMessaging_PublishClient +} + +func (mc *MessagingClient) NewPubChannel(chanName string) (*PubChannel, error) { + tp := broker.TopicPartition{ + Namespace: "chan", + Topic: chanName, + Partition: 0, + } + grpcConnection, err := mc.findBroker(tp) + if err != nil { + return nil, err + } + pc, err := setupPublisherClient(grpcConnection, tp) + if err != nil { + return nil, err + } + return &PubChannel{ + client: pc, + }, nil +} + +func (pc *PubChannel) Publish(m []byte) error { + return pc.client.Send(&messaging_pb.PublishRequest{ + Data: &messaging_pb.Message{ + Value: m, + }, + }) +} +func (pc *PubChannel) Close() error { + return pc.client.CloseSend() +} + +type SubChannel struct { + ch chan []byte + stream messaging_pb.SeaweedMessaging_SubscribeClient +} + +func (mc *MessagingClient) NewSubChannel(chanName string) (*SubChannel, error) { + tp := broker.TopicPartition{ + Namespace: "chan", + Topic: chanName, + Partition: 0, + } + grpcConnection, err := mc.findBroker(tp) + if err != nil { + return nil, err + } + sc, err := setupSubscriberClient(grpcConnection, "", "chan", chanName, 0, time.Unix(0,0)) + if err != nil { + return nil, err + } + + t := &SubChannel{ + ch: make(chan []byte), + stream: sc, + } + + go func() { + for { + resp, subErr := t.stream.Recv() + if subErr == io.EOF { + return + } + if subErr != nil { + log.Printf("fail to receive from netchan %s: %v", chanName, subErr) + return + } + if resp.IsClose { + close(t.ch) + return + } + if resp.Data != nil { + t.ch <- resp.Data.Value + } + } + }() + + return t, nil +} + +func (sc *SubChannel) Channel() chan []byte { + return sc.ch +} diff --git a/weed/messaging/client/publisher.go b/weed/messaging/msgclient/publisher.go similarity index 84% rename from weed/messaging/client/publisher.go rename to weed/messaging/msgclient/publisher.go index 68e5729c1..b0459494b 100644 --- a/weed/messaging/client/publisher.go +++ b/weed/messaging/msgclient/publisher.go @@ -1,10 +1,12 @@ -package client +package msgclient import ( "context" "github.com/OneOfOne/xxhash" + "github.com/chrislusf/seaweedfs/weed/messaging/broker" + "github.com/chrislusf/seaweedfs/weed/pb" "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb" ) @@ -34,9 +36,9 @@ func (mc *MessagingClient) NewPublisher(publisherId, namespace, topic string) (* }, nil } -func (mc *MessagingClient) setupPublisherClient(namespace, topic string, partition int32) (messaging_pb.SeaweedMessaging_PublishClient, error) { +func setupPublisherClient(grpcConnection *grpc.ClientConn, tp broker.TopicPartition) (messaging_pb.SeaweedMessaging_PublishClient, error) { - stream, err := messaging_pb.NewSeaweedMessagingClient(mc.grpcConnection).Publish(context.Background()) + stream, err := messaging_pb.NewSeaweedMessagingClient(grpcConnection).Publish(context.Background()) if err != nil { return nil, err } @@ -44,9 +46,9 @@ func (mc *MessagingClient) setupPublisherClient(namespace, topic string, partiti // send init message err = stream.Send(&messaging_pb.PublishRequest{ Init: &messaging_pb.PublishRequest_InitMessage{ - Namespace: namespace, - Topic: topic, - Partition: partition, + Namespace: tp.Namespace, + Topic: tp.Topic, + Partition: tp.Partition, }, }) if err != nil { diff --git a/weed/messaging/client/subscriber.go b/weed/messaging/msgclient/subscriber.go similarity index 79% rename from weed/messaging/client/subscriber.go rename to weed/messaging/msgclient/subscriber.go index 53e7ffc7d..27fa35a5b 100644 --- a/weed/messaging/client/subscriber.go +++ b/weed/messaging/msgclient/subscriber.go @@ -1,4 +1,4 @@ -package client +package msgclient import ( "context" @@ -36,9 +36,22 @@ func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string, func (mc *MessagingClient) setupSubscriberClient(subscriberId, namespace, topic string, partition int32, startTime time.Time) (messaging_pb.SeaweedMessaging_SubscribeClient, error) { - stream, err := messaging_pb.NewSeaweedMessagingClient(mc.grpcConnection).Subscribe(context.Background()) + stream, newBroker, err := mc.initSubscriberClient(subscriberId, namespace, topic, partition, startTime) if err != nil { - return nil, err + return client, err + } + if newBroker != nil { + + } + + return stream, nil + +} + +func setupSubscriberClient(grpcConnection *grpc.ClientConn, subscriberId string, namespace string, topic string, partition int32, startTime time.Time) (stream messaging_pb.SeaweedMessaging_SubscribeClient, err error) { + stream, err = messaging_pb.NewSeaweedMessagingClient(grpcConnection).Subscribe(context.Background()) + if err != nil { + return } // send init message @@ -53,20 +66,18 @@ func (mc *MessagingClient) setupSubscriberClient(subscriberId, namespace, topic }, }) if err != nil { - return nil, err + return } // process init response initResponse, err := stream.Recv() if err != nil { - return nil, err + return } if initResponse.Redirect != nil { // TODO follow redirection } - return stream, nil - } func (s *Subscriber) doSubscribe(partition int, processFn func(m *messaging_pb.Message)) error { diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 6d890861d..3b3b78bbb 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -48,6 +48,8 @@ service SeaweedFiler { rpc SubscribeMetadata (SubscribeMetadataRequest) returns (stream SubscribeMetadataResponse) { } + rpc KeepConnected (stream KeepConnectedRequest) returns (stream KeepConnectedResponse) { + } } ////////////////////////////////////////////////// @@ -261,3 +263,10 @@ message LogEntry { int32 partition_key_hash = 2; bytes data = 3; } + +message KeepConnectedRequest { + string name = 1; + uint32 grpc_port = 2; +} +message KeepConnectedResponse { +} diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 83f0eae0d..f1320273f 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -44,6 +44,8 @@ It has these top-level messages: SubscribeMetadataRequest SubscribeMetadataResponse LogEntry + KeepConnectedRequest + KeepConnectedResponse */ package filer_pb @@ -1228,6 +1230,38 @@ func (m *LogEntry) GetData() []byte { return nil } +type KeepConnectedRequest struct { + Name string `protobuf:"bytes,1,opt,name=name" json:"name,omitempty"` + GrpcPort uint32 `protobuf:"varint,2,opt,name=grpc_port,json=grpcPort" json:"grpc_port,omitempty"` +} + +func (m *KeepConnectedRequest) Reset() { *m = KeepConnectedRequest{} } +func (m *KeepConnectedRequest) String() string { return proto.CompactTextString(m) } +func (*KeepConnectedRequest) ProtoMessage() {} +func (*KeepConnectedRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{35} } + +func (m *KeepConnectedRequest) GetName() string { + if m != nil { + return m.Name + } + return "" +} + +func (m *KeepConnectedRequest) GetGrpcPort() uint32 { + if m != nil { + return m.GrpcPort + } + return 0 +} + +type KeepConnectedResponse struct { +} + +func (m *KeepConnectedResponse) Reset() { *m = KeepConnectedResponse{} } +func (m *KeepConnectedResponse) String() string { return proto.CompactTextString(m) } +func (*KeepConnectedResponse) ProtoMessage() {} +func (*KeepConnectedResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{36} } + func init() { proto.RegisterType((*LookupDirectoryEntryRequest)(nil), "filer_pb.LookupDirectoryEntryRequest") proto.RegisterType((*LookupDirectoryEntryResponse)(nil), "filer_pb.LookupDirectoryEntryResponse") @@ -1264,6 +1298,8 @@ func init() { proto.RegisterType((*SubscribeMetadataRequest)(nil), "filer_pb.SubscribeMetadataRequest") proto.RegisterType((*SubscribeMetadataResponse)(nil), "filer_pb.SubscribeMetadataResponse") proto.RegisterType((*LogEntry)(nil), "filer_pb.LogEntry") + proto.RegisterType((*KeepConnectedRequest)(nil), "filer_pb.KeepConnectedRequest") + proto.RegisterType((*KeepConnectedResponse)(nil), "filer_pb.KeepConnectedResponse") } // Reference imports to suppress errors if they are not otherwise used. @@ -1290,6 +1326,7 @@ type SeaweedFilerClient interface { Statistics(ctx context.Context, in *StatisticsRequest, opts ...grpc.CallOption) (*StatisticsResponse, error) GetFilerConfiguration(ctx context.Context, in *GetFilerConfigurationRequest, opts ...grpc.CallOption) (*GetFilerConfigurationResponse, error) SubscribeMetadata(ctx context.Context, in *SubscribeMetadataRequest, opts ...grpc.CallOption) (SeaweedFiler_SubscribeMetadataClient, error) + KeepConnected(ctx context.Context, opts ...grpc.CallOption) (SeaweedFiler_KeepConnectedClient, error) } type seaweedFilerClient struct { @@ -1463,6 +1500,37 @@ func (x *seaweedFilerSubscribeMetadataClient) Recv() (*SubscribeMetadataResponse return m, nil } +func (c *seaweedFilerClient) KeepConnected(ctx context.Context, opts ...grpc.CallOption) (SeaweedFiler_KeepConnectedClient, error) { + stream, err := grpc.NewClientStream(ctx, &_SeaweedFiler_serviceDesc.Streams[2], c.cc, "/filer_pb.SeaweedFiler/KeepConnected", opts...) + if err != nil { + return nil, err + } + x := &seaweedFilerKeepConnectedClient{stream} + return x, nil +} + +type SeaweedFiler_KeepConnectedClient interface { + Send(*KeepConnectedRequest) error + Recv() (*KeepConnectedResponse, error) + grpc.ClientStream +} + +type seaweedFilerKeepConnectedClient struct { + grpc.ClientStream +} + +func (x *seaweedFilerKeepConnectedClient) Send(m *KeepConnectedRequest) error { + return x.ClientStream.SendMsg(m) +} + +func (x *seaweedFilerKeepConnectedClient) Recv() (*KeepConnectedResponse, error) { + m := new(KeepConnectedResponse) + if err := x.ClientStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + // Server API for SeaweedFiler service type SeaweedFilerServer interface { @@ -1479,6 +1547,7 @@ type SeaweedFilerServer interface { Statistics(context.Context, *StatisticsRequest) (*StatisticsResponse, error) GetFilerConfiguration(context.Context, *GetFilerConfigurationRequest) (*GetFilerConfigurationResponse, error) SubscribeMetadata(*SubscribeMetadataRequest, SeaweedFiler_SubscribeMetadataServer) error + KeepConnected(SeaweedFiler_KeepConnectedServer) error } func RegisterSeaweedFilerServer(s *grpc.Server, srv SeaweedFilerServer) { @@ -1725,6 +1794,32 @@ func (x *seaweedFilerSubscribeMetadataServer) Send(m *SubscribeMetadataResponse) return x.ServerStream.SendMsg(m) } +func _SeaweedFiler_KeepConnected_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(SeaweedFilerServer).KeepConnected(&seaweedFilerKeepConnectedServer{stream}) +} + +type SeaweedFiler_KeepConnectedServer interface { + Send(*KeepConnectedResponse) error + Recv() (*KeepConnectedRequest, error) + grpc.ServerStream +} + +type seaweedFilerKeepConnectedServer struct { + grpc.ServerStream +} + +func (x *seaweedFilerKeepConnectedServer) Send(m *KeepConnectedResponse) error { + return x.ServerStream.SendMsg(m) +} + +func (x *seaweedFilerKeepConnectedServer) Recv() (*KeepConnectedRequest, error) { + m := new(KeepConnectedRequest) + if err := x.ServerStream.RecvMsg(m); err != nil { + return nil, err + } + return m, nil +} + var _SeaweedFiler_serviceDesc = grpc.ServiceDesc{ ServiceName: "filer_pb.SeaweedFiler", HandlerType: (*SeaweedFilerServer)(nil), @@ -1785,6 +1880,12 @@ var _SeaweedFiler_serviceDesc = grpc.ServiceDesc{ Handler: _SeaweedFiler_SubscribeMetadata_Handler, ServerStreams: true, }, + { + StreamName: "KeepConnected", + Handler: _SeaweedFiler_KeepConnected_Handler, + ServerStreams: true, + ClientStreams: true, + }, }, Metadata: "filer.proto", } @@ -1792,128 +1893,132 @@ var _SeaweedFiler_serviceDesc = grpc.ServiceDesc{ func init() { proto.RegisterFile("filer.proto", fileDescriptor0) } var fileDescriptor0 = []byte{ - // 1957 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x58, 0x5f, 0x6f, 0xdb, 0xc8, - 0x11, 0x0f, 0x25, 0x4b, 0x16, 0x47, 0x52, 0xce, 0x5e, 0xdb, 0x89, 0xac, 0xd8, 0x8e, 0x8f, 0x69, - 0xae, 0x29, 0x12, 0xb8, 0x81, 0x7b, 0x05, 0xee, 0x7a, 0xed, 0x43, 0xe2, 0x38, 0xd7, 0xf4, 0x12, - 0x5f, 0x40, 0x27, 0x45, 0x8b, 0x02, 0x65, 0x69, 0x72, 0x2d, 0x6d, 0x4d, 0x91, 0xec, 0xee, 0xd2, - 0x7f, 0xee, 0xe9, 0x3e, 0x47, 0x81, 0x7e, 0x8b, 0x3e, 0x16, 0x7d, 0x29, 0x0a, 0x14, 0xe8, 0xb7, - 0xe8, 0xf7, 0x28, 0x50, 0xec, 0x2c, 0x49, 0x2d, 0x45, 0xc9, 0xbe, 0xa0, 0xb8, 0xb7, 0xdd, 0x99, - 0xd9, 0xd9, 0xd9, 0xf9, 0xf3, 0x9b, 0x21, 0xa1, 0x7b, 0xca, 0x22, 0xca, 0xf7, 0x52, 0x9e, 0xc8, - 0x84, 0x74, 0x70, 0xe3, 0xa5, 0x27, 0xce, 0xd7, 0x70, 0xef, 0x75, 0x92, 0x9c, 0x65, 0xe9, 0x0b, - 0xc6, 0x69, 0x20, 0x13, 0x7e, 0x75, 0x18, 0x4b, 0x7e, 0xe5, 0xd2, 0x3f, 0x65, 0x54, 0x48, 0xb2, - 0x05, 0x76, 0x58, 0x30, 0x06, 0xd6, 0xae, 0xf5, 0xc8, 0x76, 0xa7, 0x04, 0x42, 0x60, 0x29, 0xf6, - 0x27, 0x74, 0xd0, 0x40, 0x06, 0xae, 0x9d, 0x43, 0xd8, 0x9a, 0xaf, 0x50, 0xa4, 0x49, 0x2c, 0x28, - 0x79, 0x08, 0x2d, 0xaa, 0x08, 0xa8, 0xad, 0xbb, 0xff, 0xd1, 0x5e, 0x61, 0xca, 0x9e, 0x96, 0xd3, - 0x5c, 0xe7, 0xef, 0x16, 0x90, 0xd7, 0x4c, 0x48, 0x45, 0x64, 0x54, 0x7c, 0x37, 0x7b, 0xee, 0x40, - 0x3b, 0xe5, 0xf4, 0x94, 0x5d, 0xe6, 0x16, 0xe5, 0x3b, 0xf2, 0x04, 0x56, 0x85, 0xf4, 0xb9, 0x7c, - 0xc9, 0x93, 0xc9, 0x4b, 0x16, 0xd1, 0x23, 0x65, 0x74, 0x13, 0x45, 0xea, 0x0c, 0xb2, 0x07, 0x84, - 0xc5, 0x41, 0x94, 0x09, 0x76, 0x4e, 0x8f, 0x0b, 0xee, 0x60, 0x69, 0xd7, 0x7a, 0xd4, 0x71, 0xe7, - 0x70, 0xc8, 0x3a, 0xb4, 0x22, 0x36, 0x61, 0x72, 0xd0, 0xda, 0xb5, 0x1e, 0xf5, 0x5d, 0xbd, 0x71, - 0x7e, 0x0e, 0x6b, 0x15, 0xfb, 0x3f, 0xec, 0xf9, 0x7f, 0x69, 0x40, 0x0b, 0x09, 0xa5, 0x8f, 0xad, - 0xa9, 0x8f, 0xc9, 0xc7, 0xd0, 0x63, 0xc2, 0x9b, 0x3a, 0xa2, 0x81, 0xb6, 0x75, 0x99, 0x28, 0x7d, - 0x4e, 0x1e, 0x43, 0x3b, 0x18, 0x67, 0xf1, 0x99, 0x18, 0x34, 0x77, 0x9b, 0x8f, 0xba, 0xfb, 0x6b, - 0xd3, 0x8b, 0xd4, 0x43, 0x0f, 0x14, 0xcf, 0xcd, 0x45, 0xc8, 0x67, 0x00, 0xbe, 0x94, 0x9c, 0x9d, - 0x64, 0x92, 0x0a, 0x7c, 0x69, 0x77, 0x7f, 0x60, 0x1c, 0xc8, 0x04, 0x7d, 0x56, 0xf2, 0x5d, 0x43, - 0x96, 0x7c, 0x0e, 0x1d, 0x7a, 0x29, 0x69, 0x1c, 0xd2, 0x70, 0xd0, 0xc2, 0x8b, 0xb6, 0x67, 0x5e, - 0xb4, 0x77, 0x98, 0xf3, 0xf5, 0xfb, 0x4a, 0xf1, 0xe1, 0x17, 0xd0, 0xaf, 0xb0, 0xc8, 0x0a, 0x34, - 0xcf, 0x68, 0x11, 0x55, 0xb5, 0x54, 0x9e, 0x3d, 0xf7, 0xa3, 0x4c, 0x27, 0x58, 0xcf, 0xd5, 0x9b, - 0x9f, 0x35, 0x3e, 0xb3, 0x9c, 0x17, 0x60, 0xbf, 0xcc, 0xa2, 0xa8, 0x3c, 0x18, 0x32, 0x5e, 0x1c, - 0x0c, 0x19, 0x9f, 0x7a, 0xb9, 0x71, 0xad, 0x97, 0xff, 0x66, 0xc1, 0xea, 0xe1, 0x39, 0x8d, 0xe5, - 0x51, 0x22, 0xd9, 0x29, 0x0b, 0x7c, 0xc9, 0x92, 0x98, 0x3c, 0x01, 0x3b, 0x89, 0x42, 0xef, 0xda, - 0x30, 0x75, 0x92, 0x28, 0xb7, 0xfa, 0x09, 0xd8, 0x31, 0xbd, 0xf0, 0xae, 0xbd, 0xae, 0x13, 0xd3, - 0x0b, 0x2d, 0xfd, 0x00, 0xfa, 0x21, 0x8d, 0xa8, 0xa4, 0x5e, 0x19, 0x1d, 0x15, 0xba, 0x9e, 0x26, - 0x1e, 0xe8, 0x70, 0x7c, 0x02, 0x1f, 0x29, 0x95, 0xa9, 0xcf, 0x69, 0x2c, 0xbd, 0xd4, 0x97, 0x63, - 0x8c, 0x89, 0xed, 0xf6, 0x63, 0x7a, 0xf1, 0x16, 0xa9, 0x6f, 0x7d, 0x39, 0x76, 0xfe, 0xda, 0x00, - 0xbb, 0x0c, 0x26, 0xb9, 0x0b, 0xcb, 0xea, 0x5a, 0x8f, 0x85, 0xb9, 0x27, 0xda, 0x6a, 0xfb, 0x2a, - 0x54, 0x55, 0x91, 0x9c, 0x9e, 0x0a, 0x2a, 0xd1, 0xbc, 0xa6, 0x9b, 0xef, 0x54, 0x66, 0x09, 0xf6, - 0x8d, 0x2e, 0x84, 0x25, 0x17, 0xd7, 0xca, 0xe3, 0x13, 0xc9, 0x26, 0x14, 0x2f, 0x6c, 0xba, 0x7a, - 0x43, 0xd6, 0xa0, 0x45, 0x3d, 0xe9, 0x8f, 0x30, 0xc3, 0x6d, 0x77, 0x89, 0xbe, 0xf3, 0x47, 0xe4, - 0x07, 0x70, 0x5b, 0x24, 0x19, 0x0f, 0xa8, 0x57, 0x5c, 0xdb, 0x46, 0x6e, 0x4f, 0x53, 0x5f, 0xea, - 0xcb, 0x1d, 0x68, 0x9e, 0xb2, 0x70, 0xb0, 0x8c, 0x8e, 0x59, 0xa9, 0x26, 0xe1, 0xab, 0xd0, 0x55, - 0x4c, 0xf2, 0x63, 0x80, 0x52, 0x53, 0x38, 0xe8, 0x2c, 0x10, 0xb5, 0x0b, 0xbd, 0x21, 0xd9, 0x06, - 0x08, 0x58, 0x3a, 0xa6, 0xdc, 0x53, 0x09, 0x63, 0x63, 0x72, 0xd8, 0x9a, 0xf2, 0x15, 0xbd, 0x52, - 0x6c, 0x26, 0xbc, 0xd1, 0x37, 0x2c, 0x4d, 0x69, 0x38, 0x00, 0xf4, 0xb0, 0xcd, 0xc4, 0x97, 0x9a, - 0xe0, 0xfc, 0x06, 0xda, 0xb9, 0x71, 0xf7, 0xc0, 0x3e, 0x4f, 0xa2, 0x6c, 0x52, 0x3a, 0xad, 0xef, - 0x76, 0x34, 0xe1, 0x55, 0x48, 0x36, 0x01, 0x51, 0x12, 0xaf, 0x68, 0xa0, 0x8b, 0xd0, 0xbf, 0xea, - 0x82, 0x3b, 0xd0, 0x0e, 0x92, 0xe4, 0x8c, 0x69, 0xdf, 0x2d, 0xbb, 0xf9, 0xce, 0xf9, 0xb6, 0x09, - 0xb7, 0xab, 0xc5, 0xa2, 0xae, 0x40, 0x2d, 0xe8, 0x69, 0x0b, 0xd5, 0xa0, 0xda, 0xe3, 0x8a, 0xb7, - 0x1b, 0xa6, 0xb7, 0x8b, 0x23, 0x93, 0x24, 0xd4, 0x17, 0xf4, 0xf5, 0x91, 0x37, 0x49, 0x48, 0x55, - 0xae, 0x67, 0x2c, 0xc4, 0xf0, 0xf4, 0x5d, 0xb5, 0x54, 0x94, 0x11, 0x0b, 0x73, 0xf0, 0x51, 0x4b, - 0x34, 0x8f, 0xa3, 0xde, 0xb6, 0x0e, 0xb8, 0xde, 0xa9, 0x80, 0x4f, 0x14, 0x75, 0x59, 0x47, 0x51, - 0xad, 0xc9, 0x2e, 0x74, 0x39, 0x4d, 0xa3, 0x3c, 0xf7, 0xd1, 0xf9, 0xb6, 0x6b, 0x92, 0xc8, 0x0e, - 0x40, 0x90, 0x44, 0x11, 0x0d, 0x50, 0xc0, 0x46, 0x01, 0x83, 0xa2, 0xf2, 0x4e, 0xca, 0xc8, 0x13, - 0x34, 0x40, 0x57, 0xb7, 0xdc, 0xb6, 0x94, 0xd1, 0x31, 0x0d, 0xd4, 0x3b, 0x32, 0x41, 0xb9, 0x87, - 0xf0, 0xd5, 0xc5, 0x73, 0x1d, 0x45, 0x40, 0x90, 0xdd, 0x06, 0x18, 0xf1, 0x24, 0x4b, 0x35, 0xb7, - 0xb7, 0xdb, 0x54, 0x48, 0x8e, 0x14, 0x64, 0x3f, 0x84, 0xdb, 0xe2, 0x6a, 0x12, 0xb1, 0xf8, 0xcc, - 0x93, 0x3e, 0x1f, 0x51, 0x39, 0xe8, 0xeb, 0x0a, 0xc8, 0xa9, 0xef, 0x90, 0xa8, 0xde, 0x3e, 0x09, - 0x7f, 0x3a, 0xb8, 0x8d, 0x19, 0xa0, 0x96, 0x4e, 0x0a, 0xe4, 0x80, 0x53, 0x5f, 0xd2, 0x0f, 0x68, - 0x63, 0xdf, 0x0d, 0x2d, 0xc8, 0x06, 0xb4, 0x13, 0x8f, 0x5e, 0x06, 0x51, 0x5e, 0xb4, 0xad, 0xe4, - 0xf0, 0x32, 0x88, 0x9c, 0xc7, 0xb0, 0x56, 0xb9, 0x31, 0x07, 0xfa, 0x75, 0x68, 0x51, 0xce, 0x93, - 0x02, 0x96, 0xf4, 0xc6, 0xf9, 0x2d, 0x90, 0xf7, 0x69, 0xf8, 0x7d, 0x98, 0xe7, 0x6c, 0xc0, 0x5a, - 0x45, 0xb5, 0xb6, 0xc3, 0xf9, 0xd6, 0x82, 0xf5, 0x67, 0x69, 0x4a, 0xe3, 0xf0, 0x5d, 0xf2, 0x01, - 0x97, 0x6e, 0x03, 0xa0, 0x5a, 0xcf, 0x68, 0xf0, 0x36, 0x52, 0x30, 0x3e, 0x1f, 0xd2, 0x5e, 0x9c, - 0xbb, 0xb0, 0x31, 0x63, 0x41, 0x6e, 0xdb, 0x3f, 0x2d, 0x20, 0x2f, 0x10, 0xf9, 0xfe, 0xbf, 0xa1, - 0x43, 0x61, 0x91, 0x6a, 0x88, 0x1a, 0x59, 0x43, 0x5f, 0xfa, 0x79, 0xbb, 0xee, 0x31, 0xa1, 0xf5, - 0xbf, 0xf0, 0xa5, 0x9f, 0xb7, 0x4d, 0x4e, 0x83, 0x8c, 0xab, 0x0e, 0x8e, 0x25, 0x83, 0x6d, 0xd3, - 0x2d, 0x48, 0xe4, 0x53, 0xb8, 0xc3, 0x46, 0x71, 0xc2, 0xe9, 0x54, 0xcc, 0xd3, 0x61, 0x6c, 0xa3, - 0xf0, 0xba, 0xe6, 0x96, 0x07, 0x0e, 0x31, 0xaa, 0x8f, 0x61, 0xad, 0xf2, 0x8c, 0x6b, 0x53, 0xe0, - 0xcf, 0x16, 0x0c, 0x9e, 0xc9, 0x64, 0xc2, 0x02, 0x97, 0x2a, 0xe3, 0x2b, 0x4f, 0x7f, 0x00, 0x7d, - 0xd5, 0x7b, 0x66, 0x9f, 0xdf, 0x4b, 0xa2, 0x70, 0xda, 0xdb, 0x37, 0x41, 0xb5, 0x1f, 0x33, 0x32, - 0xcb, 0x49, 0x14, 0x62, 0x5c, 0x1e, 0x80, 0xea, 0x11, 0xc6, 0x79, 0x3d, 0xe5, 0xf4, 0x62, 0x7a, - 0x51, 0x39, 0xaf, 0x84, 0xf0, 0xbc, 0x6e, 0x2c, 0xcb, 0x31, 0xbd, 0x50, 0xe7, 0x9d, 0x7b, 0xb0, - 0x39, 0xc7, 0xb6, 0x3c, 0x5c, 0xff, 0xb2, 0x60, 0xed, 0x99, 0x10, 0x6c, 0x14, 0xff, 0x1a, 0x41, - 0xb2, 0x30, 0x7a, 0x1d, 0x5a, 0x41, 0x92, 0xc5, 0x12, 0x8d, 0x6d, 0xb9, 0x7a, 0x33, 0x83, 0x1b, - 0x8d, 0x1a, 0x6e, 0xcc, 0x20, 0x4f, 0xb3, 0x8e, 0x3c, 0x06, 0xb2, 0x2c, 0x55, 0x90, 0xe5, 0x3e, - 0x74, 0x55, 0x90, 0xbd, 0x80, 0xc6, 0x92, 0xf2, 0xbc, 0x2b, 0x81, 0x22, 0x1d, 0x20, 0x45, 0x09, - 0x98, 0xdd, 0x53, 0x37, 0x26, 0x48, 0xa7, 0xad, 0xf3, 0x3f, 0xaa, 0x2a, 0x2a, 0x4f, 0xc9, 0x63, - 0xb6, 0xb0, 0x8b, 0x2a, 0xe0, 0xe5, 0x51, 0xfe, 0x0e, 0xb5, 0x54, 0x25, 0x92, 0x66, 0x27, 0x11, - 0x0b, 0x3c, 0xc5, 0xd0, 0xf6, 0xdb, 0x9a, 0xf2, 0x9e, 0x47, 0x53, 0xaf, 0x2c, 0x99, 0x5e, 0x21, - 0xb0, 0xe4, 0x67, 0x72, 0x5c, 0x74, 0x52, 0xb5, 0x9e, 0xf1, 0x54, 0xfb, 0x26, 0x4f, 0x2d, 0xd7, - 0x3d, 0x55, 0x66, 0x5a, 0xc7, 0xcc, 0xb4, 0x4f, 0x61, 0x4d, 0x8f, 0xe2, 0xd5, 0x70, 0x6d, 0x03, - 0x94, 0x5d, 0x4f, 0x0c, 0x2c, 0x0d, 0xbd, 0x45, 0xdb, 0x13, 0xce, 0x2f, 0xc0, 0x7e, 0x9d, 0x68, - 0xbd, 0x82, 0x3c, 0x05, 0x3b, 0x2a, 0x36, 0x28, 0xda, 0xdd, 0x27, 0xd3, 0x52, 0x2f, 0xe4, 0xdc, - 0xa9, 0x90, 0xf3, 0x05, 0x74, 0x0a, 0x72, 0xe1, 0x33, 0x6b, 0x91, 0xcf, 0x1a, 0x33, 0x3e, 0x73, - 0xfe, 0x61, 0xc1, 0x7a, 0xd5, 0xe4, 0x3c, 0x2c, 0xef, 0xa1, 0x5f, 0x5e, 0xe1, 0x4d, 0xfc, 0x34, - 0xb7, 0xe5, 0xa9, 0x69, 0x4b, 0xfd, 0x58, 0x69, 0xa0, 0x78, 0xe3, 0xa7, 0x3a, 0x97, 0x7b, 0x91, - 0x41, 0x1a, 0xbe, 0x83, 0xd5, 0x9a, 0xc8, 0x9c, 0x39, 0xf4, 0x47, 0xe6, 0x1c, 0x5a, 0x01, 0xbb, - 0xf2, 0xb4, 0x39, 0x9c, 0x7e, 0x0e, 0x77, 0x35, 0x1c, 0x1c, 0x94, 0x31, 0x2c, 0x7c, 0x5f, 0x0d, - 0xb5, 0x35, 0x1b, 0x6a, 0x67, 0x08, 0x83, 0xfa, 0xd1, 0xbc, 0xfc, 0x46, 0xb0, 0x7a, 0x2c, 0x7d, - 0xc9, 0x84, 0x64, 0x41, 0xf9, 0x41, 0x34, 0x93, 0x1b, 0xd6, 0x4d, 0xfd, 0xbb, 0x5e, 0x87, 0x2b, - 0xd0, 0x94, 0xb2, 0xc8, 0x5f, 0xb5, 0x54, 0x51, 0x20, 0xe6, 0x4d, 0x79, 0x0c, 0xbe, 0x87, 0xab, - 0x54, 0x3e, 0xc8, 0x44, 0xfa, 0x91, 0x9e, 0x8f, 0x96, 0x70, 0x3e, 0xb2, 0x91, 0x82, 0x03, 0x92, - 0x1e, 0x21, 0x42, 0xcd, 0x6d, 0xe9, 0xe9, 0x49, 0x11, 0x90, 0xb9, 0x0d, 0x80, 0xa5, 0xaa, 0xab, - 0xac, 0xad, 0xcf, 0x2a, 0xca, 0x81, 0x22, 0x38, 0x3b, 0xb0, 0xf5, 0x25, 0x95, 0xaa, 0x1b, 0xf1, - 0x83, 0x24, 0x3e, 0x65, 0xa3, 0x8c, 0xfb, 0x46, 0x28, 0x9c, 0x7f, 0x5b, 0xb0, 0xbd, 0x40, 0x20, - 0x7f, 0xf0, 0x00, 0x96, 0x27, 0xbe, 0x90, 0x94, 0x17, 0x55, 0x52, 0x6c, 0x67, 0x5d, 0xd1, 0xb8, - 0xc9, 0x15, 0xcd, 0x9a, 0x2b, 0x36, 0xa0, 0x3d, 0xf1, 0x2f, 0xbd, 0xc9, 0x49, 0x3e, 0xca, 0xb5, - 0x26, 0xfe, 0xe5, 0x9b, 0x13, 0x44, 0x36, 0xc6, 0xbd, 0x93, 0x2c, 0x38, 0xa3, 0x52, 0x94, 0xc8, - 0xc6, 0xf8, 0x73, 0x4d, 0xc1, 0xd9, 0x0e, 0x07, 0x5d, 0x84, 0x81, 0x8e, 0x9b, 0xef, 0x9c, 0x0b, - 0x18, 0x1c, 0x67, 0x27, 0x22, 0xe0, 0xec, 0x84, 0xbe, 0xa1, 0xd2, 0x57, 0x60, 0x58, 0xe4, 0xc8, - 0x7d, 0xe8, 0x06, 0x11, 0x53, 0x68, 0x68, 0x7c, 0x49, 0x82, 0x26, 0x61, 0xd7, 0x40, 0xb8, 0x94, - 0x63, 0xaf, 0xf2, 0xf1, 0x0c, 0x8a, 0xf4, 0x56, 0x7f, 0x40, 0x6f, 0x42, 0x47, 0xb0, 0x38, 0xa0, - 0x5e, 0xac, 0xbf, 0x58, 0x9a, 0xee, 0x32, 0xee, 0x8f, 0x84, 0x6a, 0x67, 0x9b, 0x73, 0x6e, 0xce, - 0x5d, 0x78, 0x7d, 0x2b, 0xff, 0x15, 0x10, 0x7a, 0x8e, 0x76, 0x19, 0xdf, 0x5f, 0x79, 0x91, 0xdd, - 0x33, 0xc6, 0x9c, 0xd9, 0x4f, 0x34, 0x77, 0x95, 0xd6, 0xbe, 0xda, 0xd6, 0xa0, 0x25, 0xc5, 0xd4, - 0xbe, 0x25, 0x29, 0x8e, 0x84, 0xe3, 0x2b, 0x30, 0x1a, 0xe9, 0xb2, 0x2e, 0x05, 0xac, 0xa9, 0x00, - 0x79, 0x02, 0x24, 0xf5, 0xb9, 0x64, 0x4a, 0x85, 0x9a, 0xf4, 0xbd, 0xb1, 0x2f, 0xc6, 0x68, 0x41, - 0xcb, 0x5d, 0x29, 0x39, 0x5f, 0xd1, 0xab, 0x5f, 0xfa, 0x62, 0xac, 0xc0, 0x1b, 0x87, 0x8b, 0x26, - 0xce, 0x9b, 0xb8, 0xde, 0xff, 0x6f, 0x07, 0x7a, 0xc7, 0xd4, 0xbf, 0xa0, 0x34, 0xc4, 0x54, 0x22, - 0xa3, 0x02, 0xc2, 0xaa, 0x3f, 0x40, 0xc8, 0xc3, 0x59, 0xac, 0x9a, 0xfb, 0xc7, 0x65, 0xf8, 0xc9, - 0x4d, 0x62, 0x39, 0x1a, 0xdc, 0x22, 0x47, 0xd0, 0x35, 0xfe, 0x30, 0x90, 0x2d, 0xe3, 0x60, 0xed, - 0xc7, 0xc9, 0x70, 0x7b, 0x01, 0xb7, 0xd0, 0xf6, 0xd4, 0x22, 0xaf, 0xa1, 0x6b, 0x0c, 0xb2, 0xa6, - 0xbe, 0xfa, 0x44, 0x6d, 0xea, 0x9b, 0x33, 0xfd, 0x3a, 0xb7, 0x94, 0x36, 0x63, 0x1c, 0x35, 0xb5, - 0xd5, 0x07, 0x60, 0x53, 0xdb, 0xbc, 0x19, 0xf6, 0x16, 0x71, 0xa1, 0x5f, 0x19, 0x21, 0xc9, 0xce, - 0xf4, 0xc4, 0xbc, 0xe9, 0x76, 0x78, 0x7f, 0x21, 0xdf, 0xb4, 0xd0, 0x98, 0xda, 0x4c, 0x0b, 0xeb, - 0x33, 0xa9, 0x69, 0xe1, 0x9c, 0x51, 0xcf, 0xb9, 0x45, 0x7e, 0x0f, 0xab, 0xb5, 0xc9, 0x89, 0x38, - 0x86, 0x15, 0x0b, 0x46, 0xbe, 0xe1, 0x83, 0x6b, 0x65, 0x4a, 0xfd, 0x5f, 0x43, 0xcf, 0x1c, 0x58, - 0x88, 0x61, 0xd0, 0x9c, 0x99, 0x6c, 0xb8, 0xb3, 0x88, 0x6d, 0x2a, 0x34, 0x7b, 0xa6, 0xa9, 0x70, - 0xce, 0xd4, 0x60, 0x2a, 0x9c, 0xd7, 0x6a, 0x9d, 0x5b, 0xe4, 0x77, 0xb0, 0x32, 0xdb, 0xbb, 0xc8, - 0xc7, 0xb3, 0x6e, 0xab, 0xb5, 0xc4, 0xa1, 0x73, 0x9d, 0x48, 0xa9, 0xfc, 0x15, 0xc0, 0xb4, 0x25, - 0x11, 0x03, 0x1c, 0x6a, 0x2d, 0x71, 0xb8, 0x35, 0x9f, 0x59, 0xaa, 0xfa, 0x23, 0x6c, 0xcc, 0xc5, - 0x7d, 0x62, 0x94, 0xde, 0x75, 0x9d, 0x63, 0xf8, 0xc3, 0x1b, 0xe5, 0xca, 0xbb, 0xfe, 0x00, 0xab, - 0x35, 0x70, 0x34, 0xb3, 0x62, 0x11, 0x66, 0x9b, 0x59, 0xb1, 0x10, 0x5d, 0x55, 0xd5, 0x3e, 0xdf, - 0x81, 0x15, 0xa1, 0xe1, 0xe7, 0x54, 0xec, 0x69, 0x4c, 0x7f, 0x0e, 0x68, 0xd3, 0x5b, 0x9e, 0xc8, - 0xe4, 0xa4, 0x8d, 0x7f, 0x7c, 0x7f, 0xf2, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0x39, 0x78, 0xfe, - 0x97, 0x00, 0x16, 0x00, 0x00, + // 2020 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x58, 0xcd, 0x6e, 0xdc, 0xc8, + 0x11, 0x16, 0x67, 0x34, 0xa3, 0x61, 0xcd, 0x8c, 0x57, 0x6a, 0x49, 0xf6, 0x68, 0xf4, 0x63, 0x2d, + 0x1d, 0x6f, 0x14, 0xd8, 0x50, 0x0c, 0x65, 0x03, 0xec, 0x66, 0x93, 0x83, 0x2d, 0xcb, 0x8e, 0x63, + 0x5b, 0x2b, 0x50, 0xf6, 0x22, 0x41, 0x80, 0x30, 0x14, 0xd9, 0x1a, 0x75, 0xc4, 0x21, 0x99, 0xee, + 0xa6, 0x7e, 0xf6, 0xb4, 0xcf, 0x11, 0x20, 0x6f, 0x91, 0x63, 0x90, 0x4b, 0x10, 0x20, 0x40, 0xce, + 0x79, 0x81, 0x3c, 0x49, 0xd0, 0xd5, 0x24, 0xa7, 0x39, 0x3f, 0xd2, 0x1a, 0x0b, 0xdf, 0xba, 0xab, + 0xaa, 0xab, 0xab, 0xeb, 0xe7, 0xab, 0x22, 0xa1, 0x7d, 0xca, 0x22, 0xca, 0x77, 0x53, 0x9e, 0xc8, + 0x84, 0xb4, 0x70, 0xe3, 0xa5, 0x27, 0xce, 0xd7, 0xb0, 0xfe, 0x26, 0x49, 0xce, 0xb3, 0xf4, 0x39, + 0xe3, 0x34, 0x90, 0x09, 0xbf, 0x3e, 0x88, 0x25, 0xbf, 0x76, 0xe9, 0x9f, 0x33, 0x2a, 0x24, 0xd9, + 0x00, 0x3b, 0x2c, 0x18, 0x3d, 0x6b, 0xdb, 0xda, 0xb1, 0xdd, 0x11, 0x81, 0x10, 0x98, 0x8f, 0xfd, + 0x21, 0xed, 0xd5, 0x90, 0x81, 0x6b, 0xe7, 0x00, 0x36, 0xa6, 0x2b, 0x14, 0x69, 0x12, 0x0b, 0x4a, + 0x1e, 0x42, 0x83, 0x2a, 0x02, 0x6a, 0x6b, 0xef, 0x7d, 0xb2, 0x5b, 0x98, 0xb2, 0xab, 0xe5, 0x34, + 0xd7, 0xf9, 0x87, 0x05, 0xe4, 0x0d, 0x13, 0x52, 0x11, 0x19, 0x15, 0xdf, 0xcf, 0x9e, 0xbb, 0xd0, + 0x4c, 0x39, 0x3d, 0x65, 0x57, 0xb9, 0x45, 0xf9, 0x8e, 0x3c, 0x86, 0x25, 0x21, 0x7d, 0x2e, 0x5f, + 0xf0, 0x64, 0xf8, 0x82, 0x45, 0xf4, 0x50, 0x19, 0x5d, 0x47, 0x91, 0x49, 0x06, 0xd9, 0x05, 0xc2, + 0xe2, 0x20, 0xca, 0x04, 0xbb, 0xa0, 0xc7, 0x05, 0xb7, 0x37, 0xbf, 0x6d, 0xed, 0xb4, 0xdc, 0x29, + 0x1c, 0xb2, 0x02, 0x8d, 0x88, 0x0d, 0x99, 0xec, 0x35, 0xb6, 0xad, 0x9d, 0xae, 0xab, 0x37, 0xce, + 0x2f, 0x61, 0xb9, 0x62, 0xff, 0x87, 0x3d, 0xff, 0xaf, 0x35, 0x68, 0x20, 0xa1, 0xf4, 0xb1, 0x35, + 0xf2, 0x31, 0xf9, 0x14, 0x3a, 0x4c, 0x78, 0x23, 0x47, 0xd4, 0xd0, 0xb6, 0x36, 0x13, 0xa5, 0xcf, + 0xc9, 0x23, 0x68, 0x06, 0x67, 0x59, 0x7c, 0x2e, 0x7a, 0xf5, 0xed, 0xfa, 0x4e, 0x7b, 0x6f, 0x79, + 0x74, 0x91, 0x7a, 0xe8, 0xbe, 0xe2, 0xb9, 0xb9, 0x08, 0xf9, 0x02, 0xc0, 0x97, 0x92, 0xb3, 0x93, + 0x4c, 0x52, 0x81, 0x2f, 0x6d, 0xef, 0xf5, 0x8c, 0x03, 0x99, 0xa0, 0x4f, 0x4b, 0xbe, 0x6b, 0xc8, + 0x92, 0x2f, 0xa1, 0x45, 0xaf, 0x24, 0x8d, 0x43, 0x1a, 0xf6, 0x1a, 0x78, 0xd1, 0xe6, 0xd8, 0x8b, + 0x76, 0x0f, 0x72, 0xbe, 0x7e, 0x5f, 0x29, 0xde, 0xff, 0x0a, 0xba, 0x15, 0x16, 0x59, 0x84, 0xfa, + 0x39, 0x2d, 0xa2, 0xaa, 0x96, 0xca, 0xb3, 0x17, 0x7e, 0x94, 0xe9, 0x04, 0xeb, 0xb8, 0x7a, 0xf3, + 0x8b, 0xda, 0x17, 0x96, 0xf3, 0x1c, 0xec, 0x17, 0x59, 0x14, 0x95, 0x07, 0x43, 0xc6, 0x8b, 0x83, + 0x21, 0xe3, 0x23, 0x2f, 0xd7, 0x6e, 0xf4, 0xf2, 0xdf, 0x2d, 0x58, 0x3a, 0xb8, 0xa0, 0xb1, 0x3c, + 0x4c, 0x24, 0x3b, 0x65, 0x81, 0x2f, 0x59, 0x12, 0x93, 0xc7, 0x60, 0x27, 0x51, 0xe8, 0xdd, 0x18, + 0xa6, 0x56, 0x12, 0xe5, 0x56, 0x3f, 0x06, 0x3b, 0xa6, 0x97, 0xde, 0x8d, 0xd7, 0xb5, 0x62, 0x7a, + 0xa9, 0xa5, 0x1f, 0x40, 0x37, 0xa4, 0x11, 0x95, 0xd4, 0x2b, 0xa3, 0xa3, 0x42, 0xd7, 0xd1, 0xc4, + 0x7d, 0x1d, 0x8e, 0xcf, 0xe0, 0x13, 0xa5, 0x32, 0xf5, 0x39, 0x8d, 0xa5, 0x97, 0xfa, 0xf2, 0x0c, + 0x63, 0x62, 0xbb, 0xdd, 0x98, 0x5e, 0x1e, 0x21, 0xf5, 0xc8, 0x97, 0x67, 0xce, 0xdf, 0x6a, 0x60, + 0x97, 0xc1, 0x24, 0xf7, 0x60, 0x41, 0x5d, 0xeb, 0xb1, 0x30, 0xf7, 0x44, 0x53, 0x6d, 0x5f, 0x85, + 0xaa, 0x2a, 0x92, 0xd3, 0x53, 0x41, 0x25, 0x9a, 0x57, 0x77, 0xf3, 0x9d, 0xca, 0x2c, 0xc1, 0xbe, + 0xd5, 0x85, 0x30, 0xef, 0xe2, 0x5a, 0x79, 0x7c, 0x28, 0xd9, 0x90, 0xe2, 0x85, 0x75, 0x57, 0x6f, + 0xc8, 0x32, 0x34, 0xa8, 0x27, 0xfd, 0x01, 0x66, 0xb8, 0xed, 0xce, 0xd3, 0x77, 0xfe, 0x80, 0xfc, + 0x08, 0xee, 0x88, 0x24, 0xe3, 0x01, 0xf5, 0x8a, 0x6b, 0x9b, 0xc8, 0xed, 0x68, 0xea, 0x0b, 0x7d, + 0xb9, 0x03, 0xf5, 0x53, 0x16, 0xf6, 0x16, 0xd0, 0x31, 0x8b, 0xd5, 0x24, 0x7c, 0x15, 0xba, 0x8a, + 0x49, 0x7e, 0x0a, 0x50, 0x6a, 0x0a, 0x7b, 0xad, 0x19, 0xa2, 0x76, 0xa1, 0x37, 0x24, 0x9b, 0x00, + 0x01, 0x4b, 0xcf, 0x28, 0xf7, 0x54, 0xc2, 0xd8, 0x98, 0x1c, 0xb6, 0xa6, 0xbc, 0xa6, 0xd7, 0x8a, + 0xcd, 0x84, 0x37, 0xf8, 0x96, 0xa5, 0x29, 0x0d, 0x7b, 0x80, 0x1e, 0xb6, 0x99, 0x78, 0xa9, 0x09, + 0xce, 0x6f, 0xa1, 0x99, 0x1b, 0xb7, 0x0e, 0xf6, 0x45, 0x12, 0x65, 0xc3, 0xd2, 0x69, 0x5d, 0xb7, + 0xa5, 0x09, 0xaf, 0x42, 0xb2, 0x06, 0x88, 0x92, 0x78, 0x45, 0x0d, 0x5d, 0x84, 0xfe, 0x55, 0x17, + 0xdc, 0x85, 0x66, 0x90, 0x24, 0xe7, 0x4c, 0xfb, 0x6e, 0xc1, 0xcd, 0x77, 0xce, 0x77, 0x75, 0xb8, + 0x53, 0x2d, 0x16, 0x75, 0x05, 0x6a, 0x41, 0x4f, 0x5b, 0xa8, 0x06, 0xd5, 0x1e, 0x57, 0xbc, 0x5d, + 0x33, 0xbd, 0x5d, 0x1c, 0x19, 0x26, 0xa1, 0xbe, 0xa0, 0xab, 0x8f, 0xbc, 0x4d, 0x42, 0xaa, 0x72, + 0x3d, 0x63, 0x21, 0x86, 0xa7, 0xeb, 0xaa, 0xa5, 0xa2, 0x0c, 0x58, 0x98, 0x83, 0x8f, 0x5a, 0xa2, + 0x79, 0x1c, 0xf5, 0x36, 0x75, 0xc0, 0xf5, 0x4e, 0x05, 0x7c, 0xa8, 0xa8, 0x0b, 0x3a, 0x8a, 0x6a, + 0x4d, 0xb6, 0xa1, 0xcd, 0x69, 0x1a, 0xe5, 0xb9, 0x8f, 0xce, 0xb7, 0x5d, 0x93, 0x44, 0xb6, 0x00, + 0x82, 0x24, 0x8a, 0x68, 0x80, 0x02, 0x36, 0x0a, 0x18, 0x14, 0x95, 0x77, 0x52, 0x46, 0x9e, 0xa0, + 0x01, 0xba, 0xba, 0xe1, 0x36, 0xa5, 0x8c, 0x8e, 0x69, 0xa0, 0xde, 0x91, 0x09, 0xca, 0x3d, 0x84, + 0xaf, 0x36, 0x9e, 0x6b, 0x29, 0x02, 0x82, 0xec, 0x26, 0xc0, 0x80, 0x27, 0x59, 0xaa, 0xb9, 0x9d, + 0xed, 0xba, 0x42, 0x72, 0xa4, 0x20, 0xfb, 0x21, 0xdc, 0x11, 0xd7, 0xc3, 0x88, 0xc5, 0xe7, 0x9e, + 0xf4, 0xf9, 0x80, 0xca, 0x5e, 0x57, 0x57, 0x40, 0x4e, 0x7d, 0x87, 0x44, 0xf5, 0xf6, 0x61, 0xf8, + 0xf3, 0xde, 0x1d, 0xcc, 0x00, 0xb5, 0x74, 0x52, 0x20, 0xfb, 0x9c, 0xfa, 0x92, 0x7e, 0x40, 0x1b, + 0xfb, 0x7e, 0x68, 0x41, 0x56, 0xa1, 0x99, 0x78, 0xf4, 0x2a, 0x88, 0xf2, 0xa2, 0x6d, 0x24, 0x07, + 0x57, 0x41, 0xe4, 0x3c, 0x82, 0xe5, 0xca, 0x8d, 0x39, 0xd0, 0xaf, 0x40, 0x83, 0x72, 0x9e, 0x14, + 0xb0, 0xa4, 0x37, 0xce, 0xef, 0x80, 0xbc, 0x4f, 0xc3, 0x8f, 0x61, 0x9e, 0xb3, 0x0a, 0xcb, 0x15, + 0xd5, 0xda, 0x0e, 0xe7, 0x3b, 0x0b, 0x56, 0x9e, 0xa6, 0x29, 0x8d, 0xc3, 0x77, 0xc9, 0x07, 0x5c, + 0xba, 0x09, 0x80, 0x6a, 0x3d, 0xa3, 0xc1, 0xdb, 0x48, 0xc1, 0xf8, 0x7c, 0x48, 0x7b, 0x71, 0xee, + 0xc1, 0xea, 0x98, 0x05, 0xb9, 0x6d, 0xff, 0xb2, 0x80, 0x3c, 0x47, 0xe4, 0xfb, 0x61, 0x43, 0x87, + 0xc2, 0x22, 0xd5, 0x10, 0x35, 0xb2, 0x86, 0xbe, 0xf4, 0xf3, 0x76, 0xdd, 0x61, 0x42, 0xeb, 0x7f, + 0xee, 0x4b, 0x3f, 0x6f, 0x9b, 0x9c, 0x06, 0x19, 0x57, 0x1d, 0x1c, 0x4b, 0x06, 0xdb, 0xa6, 0x5b, + 0x90, 0xc8, 0xe7, 0x70, 0x97, 0x0d, 0xe2, 0x84, 0xd3, 0x91, 0x98, 0xa7, 0xc3, 0xd8, 0x44, 0xe1, + 0x15, 0xcd, 0x2d, 0x0f, 0x1c, 0x60, 0x54, 0x1f, 0xc1, 0x72, 0xe5, 0x19, 0x37, 0xa6, 0xc0, 0x5f, + 0x2c, 0xe8, 0x3d, 0x95, 0xc9, 0x90, 0x05, 0x2e, 0x55, 0xc6, 0x57, 0x9e, 0xfe, 0x00, 0xba, 0xaa, + 0xf7, 0x8c, 0x3f, 0xbf, 0x93, 0x44, 0xe1, 0xa8, 0xb7, 0xaf, 0x81, 0x6a, 0x3f, 0x66, 0x64, 0x16, + 0x92, 0x28, 0xc4, 0xb8, 0x3c, 0x00, 0xd5, 0x23, 0x8c, 0xf3, 0x7a, 0xca, 0xe9, 0xc4, 0xf4, 0xb2, + 0x72, 0x5e, 0x09, 0xe1, 0x79, 0xdd, 0x58, 0x16, 0x62, 0x7a, 0xa9, 0xce, 0x3b, 0xeb, 0xb0, 0x36, + 0xc5, 0xb6, 0x3c, 0x5c, 0xff, 0xb6, 0x60, 0xf9, 0xa9, 0x10, 0x6c, 0x10, 0x7f, 0x83, 0x20, 0x59, + 0x18, 0xbd, 0x02, 0x8d, 0x20, 0xc9, 0x62, 0x89, 0xc6, 0x36, 0x5c, 0xbd, 0x19, 0xc3, 0x8d, 0xda, + 0x04, 0x6e, 0x8c, 0x21, 0x4f, 0x7d, 0x12, 0x79, 0x0c, 0x64, 0x99, 0xaf, 0x20, 0xcb, 0x7d, 0x68, + 0xab, 0x20, 0x7b, 0x01, 0x8d, 0x25, 0xe5, 0x79, 0x57, 0x02, 0x45, 0xda, 0x47, 0x8a, 0x12, 0x30, + 0xbb, 0xa7, 0x6e, 0x4c, 0x90, 0x8e, 0x5a, 0xe7, 0xff, 0x54, 0x55, 0x54, 0x9e, 0x92, 0xc7, 0x6c, + 0x66, 0x17, 0x55, 0xc0, 0xcb, 0xa3, 0xfc, 0x1d, 0x6a, 0xa9, 0x4a, 0x24, 0xcd, 0x4e, 0x22, 0x16, + 0x78, 0x8a, 0xa1, 0xed, 0xb7, 0x35, 0xe5, 0x3d, 0x8f, 0x46, 0x5e, 0x99, 0x37, 0xbd, 0x42, 0x60, + 0xde, 0xcf, 0xe4, 0x59, 0xd1, 0x49, 0xd5, 0x7a, 0xcc, 0x53, 0xcd, 0xdb, 0x3c, 0xb5, 0x30, 0xe9, + 0xa9, 0x32, 0xd3, 0x5a, 0x66, 0xa6, 0x7d, 0x0e, 0xcb, 0x7a, 0x14, 0xaf, 0x86, 0x6b, 0x13, 0xa0, + 0xec, 0x7a, 0xa2, 0x67, 0x69, 0xe8, 0x2d, 0xda, 0x9e, 0x70, 0x7e, 0x05, 0xf6, 0x9b, 0x44, 0xeb, + 0x15, 0xe4, 0x09, 0xd8, 0x51, 0xb1, 0x41, 0xd1, 0xf6, 0x1e, 0x19, 0x95, 0x7a, 0x21, 0xe7, 0x8e, + 0x84, 0x9c, 0xaf, 0xa0, 0x55, 0x90, 0x0b, 0x9f, 0x59, 0xb3, 0x7c, 0x56, 0x1b, 0xf3, 0x99, 0xf3, + 0x4f, 0x0b, 0x56, 0xaa, 0x26, 0xe7, 0x61, 0x79, 0x0f, 0xdd, 0xf2, 0x0a, 0x6f, 0xe8, 0xa7, 0xb9, + 0x2d, 0x4f, 0x4c, 0x5b, 0x26, 0x8f, 0x95, 0x06, 0x8a, 0xb7, 0x7e, 0xaa, 0x73, 0xb9, 0x13, 0x19, + 0xa4, 0xfe, 0x3b, 0x58, 0x9a, 0x10, 0x99, 0x32, 0x87, 0xfe, 0xc4, 0x9c, 0x43, 0x2b, 0x60, 0x57, + 0x9e, 0x36, 0x87, 0xd3, 0x2f, 0xe1, 0x9e, 0x86, 0x83, 0xfd, 0x32, 0x86, 0x85, 0xef, 0xab, 0xa1, + 0xb6, 0xc6, 0x43, 0xed, 0xf4, 0xa1, 0x37, 0x79, 0x34, 0x2f, 0xbf, 0x01, 0x2c, 0x1d, 0x4b, 0x5f, + 0x32, 0x21, 0x59, 0x50, 0x7e, 0x10, 0x8d, 0xe5, 0x86, 0x75, 0x5b, 0xff, 0x9e, 0xac, 0xc3, 0x45, + 0xa8, 0x4b, 0x59, 0xe4, 0xaf, 0x5a, 0xaa, 0x28, 0x10, 0xf3, 0xa6, 0x3c, 0x06, 0x1f, 0xe1, 0x2a, + 0x95, 0x0f, 0x32, 0x91, 0x7e, 0xa4, 0xe7, 0xa3, 0x79, 0x9c, 0x8f, 0x6c, 0xa4, 0xe0, 0x80, 0xa4, + 0x47, 0x88, 0x50, 0x73, 0x1b, 0x7a, 0x7a, 0x52, 0x04, 0x64, 0x6e, 0x02, 0x60, 0xa9, 0xea, 0x2a, + 0x6b, 0xea, 0xb3, 0x8a, 0xb2, 0xaf, 0x08, 0xce, 0x16, 0x6c, 0xbc, 0xa4, 0x52, 0x75, 0x23, 0xbe, + 0x9f, 0xc4, 0xa7, 0x6c, 0x90, 0x71, 0xdf, 0x08, 0x85, 0xf3, 0x1f, 0x0b, 0x36, 0x67, 0x08, 0xe4, + 0x0f, 0xee, 0xc1, 0xc2, 0xd0, 0x17, 0x92, 0xf2, 0xa2, 0x4a, 0x8a, 0xed, 0xb8, 0x2b, 0x6a, 0xb7, + 0xb9, 0xa2, 0x3e, 0xe1, 0x8a, 0x55, 0x68, 0x0e, 0xfd, 0x2b, 0x6f, 0x78, 0x92, 0x8f, 0x72, 0x8d, + 0xa1, 0x7f, 0xf5, 0xf6, 0x04, 0x91, 0x8d, 0x71, 0xef, 0x24, 0x0b, 0xce, 0xa9, 0x14, 0x25, 0xb2, + 0x31, 0xfe, 0x4c, 0x53, 0x70, 0xb6, 0xc3, 0x41, 0x17, 0x61, 0xa0, 0xe5, 0xe6, 0x3b, 0xe7, 0x12, + 0x7a, 0xc7, 0xd9, 0x89, 0x08, 0x38, 0x3b, 0xa1, 0x6f, 0xa9, 0xf4, 0x15, 0x18, 0x16, 0x39, 0x72, + 0x1f, 0xda, 0x41, 0xc4, 0x14, 0x1a, 0x1a, 0x5f, 0x92, 0xa0, 0x49, 0xd8, 0x35, 0x10, 0x2e, 0xe5, + 0x99, 0x57, 0xf9, 0x78, 0x06, 0x45, 0x3a, 0xd2, 0x1f, 0xd0, 0x6b, 0xd0, 0x12, 0x2c, 0x0e, 0xa8, + 0x17, 0xeb, 0x2f, 0x96, 0xba, 0xbb, 0x80, 0xfb, 0x43, 0xa1, 0xda, 0xd9, 0xda, 0x94, 0x9b, 0x73, + 0x17, 0xde, 0xdc, 0xca, 0x7f, 0x03, 0x84, 0x5e, 0xa0, 0x5d, 0xc6, 0xf7, 0x57, 0x5e, 0x64, 0xeb, + 0xc6, 0x98, 0x33, 0xfe, 0x89, 0xe6, 0x2e, 0xd1, 0x89, 0xaf, 0xb6, 0x65, 0x68, 0x48, 0x31, 0xb2, + 0x6f, 0x5e, 0x8a, 0x43, 0xe1, 0xf8, 0x0a, 0x8c, 0x06, 0xba, 0xac, 0x4b, 0x01, 0x6b, 0x24, 0x40, + 0x1e, 0x03, 0x49, 0x7d, 0x2e, 0x99, 0x52, 0xa1, 0x26, 0x7d, 0xef, 0xcc, 0x17, 0x67, 0x68, 0x41, + 0xc3, 0x5d, 0x2c, 0x39, 0xaf, 0xe9, 0xf5, 0xaf, 0x7d, 0x71, 0xa6, 0xc0, 0x1b, 0x87, 0x8b, 0x3a, + 0xce, 0x9b, 0xb8, 0x76, 0x5e, 0xc2, 0xca, 0x6b, 0x4a, 0xd3, 0xfd, 0x24, 0x8e, 0x69, 0x20, 0x69, + 0x58, 0x38, 0x7d, 0xda, 0x77, 0xfb, 0x3a, 0xd8, 0x03, 0x9e, 0x06, 0x5e, 0x9a, 0x70, 0xfd, 0x31, + 0xd6, 0x75, 0x5b, 0x8a, 0x70, 0x94, 0x70, 0xa9, 0xa6, 0xa4, 0x31, 0x45, 0xda, 0x87, 0x7b, 0xff, + 0xb5, 0xa1, 0x73, 0x4c, 0xfd, 0x4b, 0x4a, 0x43, 0x4c, 0x56, 0x32, 0x28, 0x40, 0xb2, 0xfa, 0x8b, + 0x85, 0x3c, 0x1c, 0x47, 0xc3, 0xa9, 0xff, 0x74, 0xfa, 0x9f, 0xdd, 0x26, 0x96, 0xe3, 0xcd, 0x1c, + 0x39, 0x84, 0xb6, 0xf1, 0x0f, 0x83, 0x6c, 0x18, 0x07, 0x27, 0x7e, 0xcd, 0xf4, 0x37, 0x67, 0x70, + 0x0b, 0x6d, 0x4f, 0x2c, 0xf2, 0x06, 0xda, 0xc6, 0xa8, 0x6c, 0xea, 0x9b, 0x9c, 0xd9, 0x4d, 0x7d, + 0x53, 0xe6, 0x6b, 0x67, 0x4e, 0x69, 0x33, 0x06, 0x5e, 0x53, 0xdb, 0xe4, 0x88, 0x6d, 0x6a, 0x9b, + 0x36, 0x25, 0xcf, 0x11, 0x17, 0xba, 0x95, 0x21, 0x95, 0x6c, 0x8d, 0x4e, 0x4c, 0x9b, 0x9f, 0xfb, + 0xf7, 0x67, 0xf2, 0x4d, 0x0b, 0x8d, 0xb9, 0xd0, 0xb4, 0x70, 0x72, 0xea, 0x35, 0x2d, 0x9c, 0x32, + 0x4c, 0x3a, 0x73, 0xe4, 0x0f, 0xb0, 0x34, 0x31, 0x9b, 0x11, 0xc7, 0xb0, 0x62, 0xc6, 0x50, 0xd9, + 0x7f, 0x70, 0xa3, 0x4c, 0xa9, 0xff, 0x6b, 0xe8, 0x98, 0x23, 0x11, 0x31, 0x0c, 0x9a, 0x32, 0xf5, + 0xf5, 0xb7, 0x66, 0xb1, 0x4d, 0x85, 0x66, 0x57, 0x36, 0x15, 0x4e, 0x99, 0x4b, 0x4c, 0x85, 0xd3, + 0x9a, 0xb9, 0x33, 0x47, 0x7e, 0x0f, 0x8b, 0xe3, 0xdd, 0x91, 0x7c, 0x3a, 0xee, 0xb6, 0x89, 0xa6, + 0xdb, 0x77, 0x6e, 0x12, 0x29, 0x95, 0xbf, 0x02, 0x18, 0x35, 0x3d, 0x62, 0xc0, 0xcf, 0x44, 0xd3, + 0xed, 0x6f, 0x4c, 0x67, 0x96, 0xaa, 0xfe, 0x04, 0xab, 0x53, 0x3b, 0x0b, 0x31, 0x4a, 0xef, 0xa6, + 0xde, 0xd4, 0xff, 0xf1, 0xad, 0x72, 0xe5, 0x5d, 0x7f, 0x84, 0xa5, 0x09, 0xf8, 0x35, 0xb3, 0x62, + 0x56, 0x57, 0x30, 0xb3, 0x62, 0x26, 0x7e, 0x63, 0xd5, 0x7e, 0x03, 0xdd, 0x0a, 0x30, 0x99, 0x95, + 0x31, 0x0d, 0xfa, 0xcc, 0xca, 0x98, 0x8a, 0x68, 0xce, 0xdc, 0x8e, 0xf5, 0xc4, 0x7a, 0xb6, 0x05, + 0x8b, 0x42, 0xc3, 0xda, 0xa9, 0xd8, 0xd5, 0xdd, 0xe8, 0x19, 0xe0, 0x5b, 0x8f, 0x78, 0x22, 0x93, + 0x93, 0x26, 0xfe, 0xab, 0xfe, 0xd9, 0xff, 0x03, 0x00, 0x00, 0xff, 0xff, 0xf8, 0xd4, 0xa1, 0xe3, + 0xba, 0x16, 0x00, 0x00, } diff --git a/weed/pb/messaging.proto b/weed/pb/messaging.proto index 165542681..56f60b212 100644 --- a/weed/pb/messaging.proto +++ b/weed/pb/messaging.proto @@ -21,6 +21,9 @@ service SeaweedMessaging { rpc GetTopicConfiguration (GetTopicConfigurationRequest) returns (GetTopicConfigurationResponse) { } + rpc FindBroker (FindBrokerRequest) returns (FindBrokerResponse) { + } + } ////////////////////////////////////////////////// @@ -55,10 +58,7 @@ message Message { message BrokerMessage { Message data = 1; - message RedirectMessage { - string new_broker = 1; - } - RedirectMessage redirect = 2; + bool is_close = 2; } message PublishRequest { @@ -98,6 +98,16 @@ message GetTopicConfigurationResponse { TopicConfiguration configuration = 1; } +message FindBrokerRequest { + string namespace = 1; + string topic = 2; + int32 parition = 3; +} + +message FindBrokerResponse { + string broker = 1; +} + message TopicConfiguration { int32 partition_count = 1; string collection = 2; diff --git a/weed/pb/messaging_pb/messaging.pb.go b/weed/pb/messaging_pb/messaging.pb.go index 57f316708..809fd3ded 100644 --- a/weed/pb/messaging_pb/messaging.pb.go +++ b/weed/pb/messaging_pb/messaging.pb.go @@ -18,6 +18,8 @@ It has these top-level messages: ConfigureTopicResponse GetTopicConfigurationRequest GetTopicConfigurationResponse + FindBrokerRequest + FindBrokerResponse TopicConfiguration */ package messaging_pb @@ -91,7 +93,7 @@ func (x TopicConfiguration_Partitioning) String() string { return proto.EnumName(TopicConfiguration_Partitioning_name, int32(x)) } func (TopicConfiguration_Partitioning) EnumDescriptor() ([]byte, []int) { - return fileDescriptor0, []int{9, 0} + return fileDescriptor0, []int{11, 0} } type SubscriberMessage struct { @@ -233,8 +235,8 @@ func (m *Message) GetHeaders() map[string][]byte { } type BrokerMessage struct { - Data *Message `protobuf:"bytes,1,opt,name=data" json:"data,omitempty"` - Redirect *BrokerMessage_RedirectMessage `protobuf:"bytes,2,opt,name=redirect" json:"redirect,omitempty"` + Data *Message `protobuf:"bytes,1,opt,name=data" json:"data,omitempty"` + IsClose bool `protobuf:"varint,2,opt,name=is_close,json=isClose" json:"is_close,omitempty"` } func (m *BrokerMessage) Reset() { *m = BrokerMessage{} } @@ -249,29 +251,11 @@ func (m *BrokerMessage) GetData() *Message { return nil } -func (m *BrokerMessage) GetRedirect() *BrokerMessage_RedirectMessage { +func (m *BrokerMessage) GetIsClose() bool { if m != nil { - return m.Redirect + return m.IsClose } - return nil -} - -type BrokerMessage_RedirectMessage struct { - NewBroker string `protobuf:"bytes,1,opt,name=new_broker,json=newBroker" json:"new_broker,omitempty"` -} - -func (m *BrokerMessage_RedirectMessage) Reset() { *m = BrokerMessage_RedirectMessage{} } -func (m *BrokerMessage_RedirectMessage) String() string { return proto.CompactTextString(m) } -func (*BrokerMessage_RedirectMessage) ProtoMessage() {} -func (*BrokerMessage_RedirectMessage) Descriptor() ([]byte, []int) { - return fileDescriptor0, []int{2, 0} -} - -func (m *BrokerMessage_RedirectMessage) GetNewBroker() string { - if m != nil { - return m.NewBroker - } - return "" + return false } type PublishRequest struct { @@ -470,6 +454,54 @@ func (m *GetTopicConfigurationResponse) GetConfiguration() *TopicConfiguration { return nil } +type FindBrokerRequest struct { + Namespace string `protobuf:"bytes,1,opt,name=namespace" json:"namespace,omitempty"` + Topic string `protobuf:"bytes,2,opt,name=topic" json:"topic,omitempty"` + Parition int32 `protobuf:"varint,3,opt,name=parition" json:"parition,omitempty"` +} + +func (m *FindBrokerRequest) Reset() { *m = FindBrokerRequest{} } +func (m *FindBrokerRequest) String() string { return proto.CompactTextString(m) } +func (*FindBrokerRequest) ProtoMessage() {} +func (*FindBrokerRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{9} } + +func (m *FindBrokerRequest) GetNamespace() string { + if m != nil { + return m.Namespace + } + return "" +} + +func (m *FindBrokerRequest) GetTopic() string { + if m != nil { + return m.Topic + } + return "" +} + +func (m *FindBrokerRequest) GetParition() int32 { + if m != nil { + return m.Parition + } + return 0 +} + +type FindBrokerResponse struct { + Broker string `protobuf:"bytes,1,opt,name=broker" json:"broker,omitempty"` +} + +func (m *FindBrokerResponse) Reset() { *m = FindBrokerResponse{} } +func (m *FindBrokerResponse) String() string { return proto.CompactTextString(m) } +func (*FindBrokerResponse) ProtoMessage() {} +func (*FindBrokerResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{10} } + +func (m *FindBrokerResponse) GetBroker() string { + if m != nil { + return m.Broker + } + return "" +} + type TopicConfiguration struct { PartitionCount int32 `protobuf:"varint,1,opt,name=partition_count,json=partitionCount" json:"partition_count,omitempty"` Collection string `protobuf:"bytes,2,opt,name=collection" json:"collection,omitempty"` @@ -481,7 +513,7 @@ type TopicConfiguration struct { func (m *TopicConfiguration) Reset() { *m = TopicConfiguration{} } func (m *TopicConfiguration) String() string { return proto.CompactTextString(m) } func (*TopicConfiguration) ProtoMessage() {} -func (*TopicConfiguration) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{9} } +func (*TopicConfiguration) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{11} } func (m *TopicConfiguration) GetPartitionCount() int32 { if m != nil { @@ -524,7 +556,6 @@ func init() { proto.RegisterType((*SubscriberMessage_AckMessage)(nil), "messaging_pb.SubscriberMessage.AckMessage") proto.RegisterType((*Message)(nil), "messaging_pb.Message") proto.RegisterType((*BrokerMessage)(nil), "messaging_pb.BrokerMessage") - proto.RegisterType((*BrokerMessage_RedirectMessage)(nil), "messaging_pb.BrokerMessage.RedirectMessage") proto.RegisterType((*PublishRequest)(nil), "messaging_pb.PublishRequest") proto.RegisterType((*PublishRequest_InitMessage)(nil), "messaging_pb.PublishRequest.InitMessage") proto.RegisterType((*PublishResponse)(nil), "messaging_pb.PublishResponse") @@ -534,6 +565,8 @@ func init() { proto.RegisterType((*ConfigureTopicResponse)(nil), "messaging_pb.ConfigureTopicResponse") proto.RegisterType((*GetTopicConfigurationRequest)(nil), "messaging_pb.GetTopicConfigurationRequest") proto.RegisterType((*GetTopicConfigurationResponse)(nil), "messaging_pb.GetTopicConfigurationResponse") + proto.RegisterType((*FindBrokerRequest)(nil), "messaging_pb.FindBrokerRequest") + proto.RegisterType((*FindBrokerResponse)(nil), "messaging_pb.FindBrokerResponse") proto.RegisterType((*TopicConfiguration)(nil), "messaging_pb.TopicConfiguration") proto.RegisterEnum("messaging_pb.SubscriberMessage_InitMessage_StartPosition", SubscriberMessage_InitMessage_StartPosition_name, SubscriberMessage_InitMessage_StartPosition_value) proto.RegisterEnum("messaging_pb.TopicConfiguration_Partitioning", TopicConfiguration_Partitioning_name, TopicConfiguration_Partitioning_value) @@ -554,6 +587,7 @@ type SeaweedMessagingClient interface { Publish(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishClient, error) ConfigureTopic(ctx context.Context, in *ConfigureTopicRequest, opts ...grpc.CallOption) (*ConfigureTopicResponse, error) GetTopicConfiguration(ctx context.Context, in *GetTopicConfigurationRequest, opts ...grpc.CallOption) (*GetTopicConfigurationResponse, error) + FindBroker(ctx context.Context, in *FindBrokerRequest, opts ...grpc.CallOption) (*FindBrokerResponse, error) } type seaweedMessagingClient struct { @@ -644,6 +678,15 @@ func (c *seaweedMessagingClient) GetTopicConfiguration(ctx context.Context, in * return out, nil } +func (c *seaweedMessagingClient) FindBroker(ctx context.Context, in *FindBrokerRequest, opts ...grpc.CallOption) (*FindBrokerResponse, error) { + out := new(FindBrokerResponse) + err := grpc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/FindBroker", in, out, c.cc, opts...) + if err != nil { + return nil, err + } + return out, nil +} + // Server API for SeaweedMessaging service type SeaweedMessagingServer interface { @@ -651,6 +694,7 @@ type SeaweedMessagingServer interface { Publish(SeaweedMessaging_PublishServer) error ConfigureTopic(context.Context, *ConfigureTopicRequest) (*ConfigureTopicResponse, error) GetTopicConfiguration(context.Context, *GetTopicConfigurationRequest) (*GetTopicConfigurationResponse, error) + FindBroker(context.Context, *FindBrokerRequest) (*FindBrokerResponse, error) } func RegisterSeaweedMessagingServer(s *grpc.Server, srv SeaweedMessagingServer) { @@ -745,6 +789,24 @@ func _SeaweedMessaging_GetTopicConfiguration_Handler(srv interface{}, ctx contex return interceptor(ctx, in, info, handler) } +func _SeaweedMessaging_FindBroker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(FindBrokerRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SeaweedMessagingServer).FindBroker(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/messaging_pb.SeaweedMessaging/FindBroker", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SeaweedMessagingServer).FindBroker(ctx, req.(*FindBrokerRequest)) + } + return interceptor(ctx, in, info, handler) +} + var _SeaweedMessaging_serviceDesc = grpc.ServiceDesc{ ServiceName: "messaging_pb.SeaweedMessaging", HandlerType: (*SeaweedMessagingServer)(nil), @@ -757,6 +819,10 @@ var _SeaweedMessaging_serviceDesc = grpc.ServiceDesc{ MethodName: "GetTopicConfiguration", Handler: _SeaweedMessaging_GetTopicConfiguration_Handler, }, + { + MethodName: "FindBroker", + Handler: _SeaweedMessaging_FindBroker_Handler, + }, }, Streams: []grpc.StreamDesc{ { @@ -778,62 +844,65 @@ var _SeaweedMessaging_serviceDesc = grpc.ServiceDesc{ func init() { proto.RegisterFile("messaging.proto", fileDescriptor0) } var fileDescriptor0 = []byte{ - // 898 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x56, 0x5d, 0x6f, 0xe3, 0x44, - 0x17, 0xee, 0x38, 0xe9, 0x47, 0x8e, 0x93, 0x34, 0xef, 0xd1, 0x5b, 0x14, 0x99, 0x16, 0x82, 0x17, - 0x41, 0xa0, 0xc2, 0xaa, 0xc2, 0x4d, 0x59, 0xad, 0xb4, 0x6a, 0xab, 0xb2, 0x1b, 0xd1, 0x76, 0xa3, - 0x49, 0x6e, 0x91, 0xe5, 0x38, 0xb3, 0xe9, 0xa8, 0xc9, 0x38, 0x78, 0x26, 0x5b, 0xf5, 0x1a, 0x6e, - 0xb9, 0xe2, 0xaf, 0xc0, 0x0f, 0xe0, 0x37, 0x70, 0xc7, 0xaf, 0x41, 0x1e, 0x7f, 0xc4, 0x4e, 0xb2, - 0xe9, 0x52, 0x89, 0x3b, 0xfb, 0xcc, 0x73, 0x9e, 0xf3, 0x9c, 0x2f, 0x8f, 0x61, 0x7f, 0xca, 0xa4, - 0xf4, 0xc6, 0x5c, 0x8c, 0x9d, 0x59, 0x18, 0xa8, 0x00, 0xab, 0x99, 0xc1, 0x9d, 0x0d, 0xed, 0x9f, - 0xcb, 0xf0, 0xbf, 0xfe, 0x7c, 0x28, 0xfd, 0x90, 0x0f, 0x59, 0x78, 0xad, 0x8f, 0x18, 0xbe, 0x84, - 0x32, 0x17, 0x5c, 0x35, 0x49, 0x8b, 0xb4, 0xcd, 0xce, 0xb1, 0x93, 0x77, 0x71, 0x56, 0xe0, 0x4e, - 0x57, 0x70, 0x95, 0x3c, 0x53, 0xed, 0x88, 0x2f, 0xa0, 0xe4, 0xf9, 0x77, 0x4d, 0x43, 0xfb, 0x7f, - 0xfd, 0x98, 0xff, 0x99, 0x7f, 0x97, 0xba, 0x47, 0x6e, 0xd6, 0x9f, 0x06, 0x98, 0x39, 0x4e, 0x3c, - 0x84, 0x8a, 0xf0, 0xa6, 0x4c, 0xce, 0x3c, 0x9f, 0x69, 0x4d, 0x15, 0xba, 0x30, 0xe0, 0xff, 0x61, - 0x5b, 0x05, 0x33, 0xee, 0xeb, 0x68, 0x15, 0x1a, 0xbf, 0x44, 0x3e, 0x33, 0x2f, 0x54, 0x5c, 0xf1, - 0x40, 0x34, 0x4b, 0x2d, 0xd2, 0xde, 0xa6, 0x0b, 0x03, 0xba, 0x50, 0x93, 0xca, 0x0b, 0x55, 0x2f, - 0x90, 0x31, 0xa2, 0xdc, 0x22, 0xed, 0x7a, 0xe7, 0xbb, 0x7f, 0x91, 0xa9, 0xd3, 0xcf, 0x13, 0xd0, - 0x22, 0x1f, 0xb6, 0xc0, 0x54, 0x7c, 0xca, 0xa4, 0xf2, 0xa6, 0xb3, 0x1b, 0xd9, 0xdc, 0x6e, 0x91, - 0x76, 0x89, 0xe6, 0x4d, 0xf8, 0x0c, 0x6a, 0x32, 0xe3, 0x77, 0xf9, 0xa8, 0xb9, 0xa3, 0xe5, 0x57, - 0x17, 0xc6, 0xee, 0xc8, 0x3e, 0x85, 0x5a, 0x21, 0x0c, 0x02, 0xec, 0x5c, 0x9d, 0x0d, 0x2e, 0xfb, - 0x83, 0xc6, 0x16, 0x56, 0x61, 0xef, 0xf2, 0x8c, 0x5e, 0x75, 0xa3, 0x37, 0x82, 0x35, 0xa8, 0x0c, - 0xba, 0xd7, 0x97, 0xfd, 0xc1, 0xd9, 0x75, 0xaf, 0x61, 0x58, 0xc7, 0x00, 0x8b, 0xb2, 0xe2, 0x11, - 0x40, 0x9c, 0x19, 0x8b, 0x22, 0x11, 0xad, 0xa6, 0x92, 0x58, 0xba, 0x23, 0xfb, 0x2f, 0x02, 0xbb, - 0x29, 0xf4, 0x0b, 0xa8, 0xb1, 0x77, 0x4c, 0x28, 0x37, 0x12, 0xeb, 0x0a, 0x19, 0xa3, 0xcf, 0x8d, - 0x13, 0x42, 0x4d, 0x7d, 0x30, 0xe0, 0x53, 0x76, 0x23, 0xb1, 0x01, 0xa5, 0x3b, 0xf6, 0xa0, 0x8b, - 0x5e, 0xa5, 0xd1, 0x63, 0xd4, 0x88, 0x77, 0xde, 0x64, 0xce, 0x74, 0xb9, 0xab, 0x34, 0x7e, 0xc1, - 0x17, 0xb0, 0x7b, 0xcb, 0xbc, 0x11, 0x0b, 0x65, 0xb3, 0xdc, 0x2a, 0xb5, 0xcd, 0x8e, 0x5d, 0x2c, - 0x72, 0x5a, 0xce, 0xd7, 0x31, 0xe8, 0x52, 0xa8, 0xf0, 0x81, 0xa6, 0x2e, 0xd6, 0x73, 0xa8, 0xe6, - 0x0f, 0xd2, 0xa8, 0xf1, 0x10, 0x14, 0xa3, 0x1a, 0xb9, 0xa8, 0xcf, 0x8d, 0x53, 0x62, 0xff, 0x41, - 0xa0, 0x76, 0x1e, 0x06, 0x77, 0x8b, 0xb9, 0xfe, 0x0a, 0xca, 0x23, 0x4f, 0x79, 0xc9, 0x5c, 0x1f, - 0xac, 0x15, 0x42, 0x35, 0x04, 0x5f, 0xc1, 0x5e, 0xc8, 0x46, 0x3c, 0x64, 0xbe, 0x4a, 0xc6, 0x78, - 0x69, 0x0d, 0x0a, 0xcc, 0x0e, 0x4d, 0xb0, 0x29, 0x49, 0xe6, 0x6c, 0x9d, 0xc0, 0xfe, 0xd2, 0x61, - 0xd4, 0x0d, 0xc1, 0xee, 0xdd, 0xa1, 0x66, 0xc8, 0x06, 0x9a, 0xdd, 0xc7, 0x94, 0xf6, 0xdf, 0x04, - 0xea, 0xbd, 0xf9, 0x70, 0xc2, 0xe5, 0x2d, 0x65, 0x3f, 0xcd, 0x99, 0x8c, 0xf6, 0x29, 0xbf, 0x90, - 0xed, 0xa2, 0x92, 0x22, 0x76, 0xcd, 0x36, 0xa6, 0x69, 0x1b, 0x8f, 0xa6, 0x6d, 0xb9, 0xff, 0xf1, - 0xe6, 0xd9, 0xbf, 0x1a, 0xb0, 0x9f, 0x09, 0x96, 0xb3, 0x40, 0x48, 0x86, 0x17, 0xb0, 0xe3, 0x07, - 0xe2, 0x2d, 0x1f, 0xaf, 0xff, 0xe0, 0x2c, 0xc1, 0x9d, 0x0b, 0x8d, 0x4d, 0x75, 0x27, 0xae, 0xd8, - 0x5d, 0x69, 0xd8, 0x37, 0x9b, 0x69, 0xde, 0xdf, 0xb2, 0x53, 0xa8, 0x15, 0x62, 0xe0, 0x97, 0xb0, - 0x9f, 0x65, 0xe0, 0xfa, 0xc1, 0x5c, 0xc4, 0x9d, 0xd8, 0xa6, 0xf5, 0xcc, 0x7c, 0x11, 0x59, 0x9f, - 0xd0, 0xec, 0xdf, 0x08, 0x1c, 0xc4, 0xc1, 0xe6, 0x21, 0x1b, 0x44, 0x05, 0x4c, 0x7b, 0xfe, 0x94, - 0xda, 0x7f, 0x0f, 0x35, 0x3f, 0x21, 0xf3, 0xb2, 0xfa, 0x9b, 0x9d, 0x56, 0xb1, 0x12, 0x3a, 0xcc, - 0x45, 0x1e, 0x47, 0x8b, 0x6e, 0x76, 0x13, 0x3e, 0x5a, 0x16, 0x15, 0x57, 0xcd, 0xa6, 0x70, 0xf8, - 0x8a, 0xa9, 0x35, 0x0c, 0x4f, 0x57, 0x6d, 0x8f, 0xe1, 0xe8, 0x3d, 0x9c, 0xc9, 0x80, 0xac, 0xa4, - 0x45, 0x9e, 0x96, 0xd6, 0xef, 0x06, 0xe0, 0x2a, 0xea, 0x83, 0xdb, 0x8b, 0x9f, 0x00, 0xf8, 0xc1, - 0x64, 0xc2, 0x7c, 0x2d, 0x22, 0xce, 0x21, 0x67, 0x89, 0xbe, 0xfa, 0x21, 0x9b, 0x4d, 0xb8, 0xbf, - 0x28, 0x7e, 0x85, 0xe6, 0x4d, 0xf8, 0x19, 0x54, 0xb9, 0x74, 0x55, 0xe8, 0x09, 0xc9, 0x99, 0x50, - 0xfa, 0xde, 0xd9, 0xa3, 0x26, 0x97, 0x83, 0xd4, 0x84, 0x6f, 0xc0, 0x8c, 0xc3, 0x06, 0x82, 0x8b, - 0xb1, 0xbe, 0x3a, 0xea, 0xcb, 0xb3, 0xbc, 0x9a, 0x84, 0xd3, 0x4b, 0xa5, 0x72, 0x31, 0xa6, 0x79, - 0x06, 0xfb, 0x25, 0x54, 0xf3, 0x87, 0x88, 0x50, 0xbf, 0x09, 0xc4, 0xcd, 0x7c, 0x32, 0xf9, 0x81, - 0x3d, 0xbc, 0xf6, 0xe4, 0x6d, 0x63, 0x0b, 0x4d, 0xd8, 0x4d, 0x5f, 0x08, 0xd6, 0x01, 0x68, 0x30, - 0x17, 0x23, 0x1a, 0x0c, 0xb9, 0x68, 0x18, 0x9d, 0x5f, 0x4a, 0xd0, 0xe8, 0x33, 0xef, 0x9e, 0xb1, - 0xd1, 0x75, 0xaa, 0x02, 0xdf, 0x40, 0x25, 0xbb, 0x1f, 0xf1, 0xd3, 0x47, 0x2e, 0x4e, 0xeb, 0xe3, - 0x0d, 0x1f, 0x4f, 0x7b, 0xab, 0x4d, 0x4e, 0x08, 0x5e, 0xc1, 0x6e, 0xb2, 0xa2, 0x78, 0xb8, 0xe9, - 0x03, 0x67, 0x1d, 0x6d, 0xdc, 0xeb, 0x84, 0xed, 0x47, 0xa8, 0x17, 0x27, 0x18, 0x9f, 0x15, 0xdd, - 0xd6, 0x2e, 0x9d, 0xf5, 0xf9, 0x66, 0x50, 0x1a, 0x02, 0x43, 0x38, 0x58, 0x3b, 0xb2, 0xb8, 0xf4, - 0xb3, 0xb3, 0x69, 0x57, 0xac, 0xe3, 0x0f, 0xc2, 0xa6, 0x31, 0xcf, 0x6d, 0x68, 0xc8, 0xb8, 0x0b, - 0x6f, 0xa5, 0xe3, 0x4f, 0xa2, 0x61, 0x39, 0xaf, 0x67, 0x0d, 0xe9, 0x45, 0x7f, 0x77, 0xc3, 0x1d, - 0xfd, 0x93, 0xf7, 0xed, 0x3f, 0x01, 0x00, 0x00, 0xff, 0xff, 0x3b, 0xcc, 0x19, 0xa5, 0xf7, 0x09, - 0x00, 0x00, + // 952 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x56, 0xdd, 0x6e, 0xe3, 0x44, + 0x14, 0xae, 0x9d, 0x34, 0x3f, 0x27, 0x3f, 0xcd, 0x1e, 0xd1, 0x55, 0x30, 0x2d, 0x04, 0x2f, 0x82, + 0x40, 0x21, 0xaa, 0xc2, 0x4d, 0x59, 0xad, 0xb4, 0x6a, 0xa3, 0x2e, 0x1b, 0xd1, 0x76, 0xc3, 0x24, + 0x5c, 0x22, 0xcb, 0xb1, 0x67, 0xd3, 0x51, 0x9d, 0xb1, 0xf1, 0x38, 0x5b, 0xf5, 0x9a, 0x6b, 0xae, + 0x78, 0x15, 0x5e, 0x80, 0x67, 0xe0, 0x02, 0x89, 0xa7, 0x41, 0x1e, 0xff, 0xc4, 0x4e, 0xb2, 0xe9, + 0x52, 0xb4, 0x77, 0x9e, 0x33, 0xdf, 0xf9, 0xce, 0x77, 0xce, 0x9c, 0x33, 0x63, 0xd8, 0x9b, 0x53, + 0x21, 0xcc, 0x19, 0xe3, 0xb3, 0x9e, 0xe7, 0xbb, 0x81, 0x8b, 0xf5, 0xd4, 0x60, 0x78, 0x53, 0xfd, + 0xd7, 0x22, 0x3c, 0x1a, 0x2f, 0xa6, 0xc2, 0xf2, 0xd9, 0x94, 0xfa, 0x97, 0x72, 0x8b, 0xe2, 0x73, + 0x28, 0x32, 0xce, 0x82, 0xb6, 0xd2, 0x51, 0xba, 0xb5, 0xfe, 0x51, 0x2f, 0xeb, 0xd2, 0x5b, 0x83, + 0xf7, 0x86, 0x9c, 0x05, 0xf1, 0x37, 0x91, 0x8e, 0xf8, 0x0c, 0x0a, 0xa6, 0x75, 0xd3, 0x56, 0xa5, + 0xff, 0x57, 0xf7, 0xf9, 0x9f, 0x5a, 0x37, 0x89, 0x7b, 0xe8, 0xa6, 0xfd, 0xa9, 0x42, 0x2d, 0xc3, + 0x89, 0x07, 0x50, 0xe5, 0xe6, 0x9c, 0x0a, 0xcf, 0xb4, 0xa8, 0xd4, 0x54, 0x25, 0x4b, 0x03, 0x7e, + 0x00, 0xbb, 0x81, 0xeb, 0x31, 0x4b, 0x46, 0xab, 0x92, 0x68, 0x11, 0xfa, 0x78, 0xa6, 0x1f, 0xb0, + 0x80, 0xb9, 0xbc, 0x5d, 0xe8, 0x28, 0xdd, 0x5d, 0xb2, 0x34, 0xa0, 0x01, 0x0d, 0x11, 0x98, 0x7e, + 0x30, 0x72, 0x45, 0x84, 0x28, 0x76, 0x94, 0x6e, 0xb3, 0xff, 0xdd, 0x7f, 0xc8, 0xb4, 0x37, 0xce, + 0x12, 0x90, 0x3c, 0x1f, 0x76, 0xa0, 0x16, 0xb0, 0x39, 0x15, 0x81, 0x39, 0xf7, 0xae, 0x44, 0x7b, + 0xb7, 0xa3, 0x74, 0x0b, 0x24, 0x6b, 0xc2, 0x27, 0xd0, 0x10, 0x29, 0xbf, 0xc1, 0xec, 0x76, 0x49, + 0xca, 0xaf, 0x2f, 0x8d, 0x43, 0x5b, 0x3f, 0x81, 0x46, 0x2e, 0x0c, 0x02, 0x94, 0x2e, 0x4e, 0x27, + 0xe7, 0xe3, 0x49, 0x6b, 0x07, 0xeb, 0x50, 0x39, 0x3f, 0x25, 0x17, 0xc3, 0x70, 0xa5, 0x60, 0x03, + 0xaa, 0x93, 0xe1, 0xe5, 0xf9, 0x78, 0x72, 0x7a, 0x39, 0x6a, 0xa9, 0xda, 0x11, 0xc0, 0xb2, 0xac, + 0x78, 0x08, 0x10, 0x65, 0x46, 0xc3, 0x48, 0x8a, 0x54, 0x53, 0x8d, 0x2d, 0x43, 0x5b, 0xff, 0x4b, + 0x81, 0x72, 0x02, 0xfd, 0x1c, 0x1a, 0xf4, 0x0d, 0xe5, 0x81, 0x11, 0x8a, 0x35, 0xb8, 0x88, 0xd0, + 0x67, 0xea, 0xb1, 0x42, 0x6a, 0x72, 0x63, 0xc2, 0xe6, 0xf4, 0x4a, 0x60, 0x0b, 0x0a, 0x37, 0xf4, + 0x4e, 0x16, 0xbd, 0x4e, 0xc2, 0xcf, 0xf0, 0x20, 0xde, 0x98, 0xce, 0x82, 0xca, 0x72, 0xd7, 0x49, + 0xb4, 0xc0, 0x67, 0x50, 0xbe, 0xa6, 0xa6, 0x4d, 0x7d, 0xd1, 0x2e, 0x76, 0x0a, 0xdd, 0x5a, 0x5f, + 0xcf, 0x17, 0x39, 0x29, 0xe7, 0xcb, 0x08, 0x74, 0xce, 0x03, 0xff, 0x8e, 0x24, 0x2e, 0xda, 0x53, + 0xa8, 0x67, 0x37, 0x92, 0xa8, 0x51, 0x13, 0xe4, 0xa3, 0xaa, 0x99, 0xa8, 0x4f, 0xd5, 0x13, 0x45, + 0xff, 0x09, 0x1a, 0x67, 0xbe, 0x7b, 0xb3, 0x6c, 0xeb, 0x2f, 0xa1, 0x68, 0x9b, 0x81, 0x19, 0xb7, + 0xf5, 0xfe, 0x46, 0x1d, 0x44, 0x42, 0xf0, 0x43, 0xa8, 0x30, 0x61, 0x58, 0x8e, 0x2b, 0x22, 0xe2, + 0x0a, 0x29, 0x33, 0x31, 0x08, 0x97, 0xfa, 0x3f, 0x0a, 0x34, 0x47, 0x8b, 0xa9, 0xc3, 0xc4, 0x35, + 0xa1, 0xbf, 0x2c, 0xa8, 0x08, 0xdb, 0x3d, 0x3b, 0x2f, 0xdd, 0x3c, 0x71, 0x1e, 0xbb, 0x61, 0x58, + 0x12, 0x59, 0xea, 0xbd, 0xb2, 0x34, 0xe3, 0x3d, 0x0f, 0x86, 0xfe, 0x9b, 0x0a, 0x7b, 0xa9, 0x60, + 0xe1, 0xb9, 0x5c, 0x50, 0x1c, 0x40, 0xc9, 0x72, 0xf9, 0x6b, 0x36, 0xdb, 0x7c, 0x1f, 0xac, 0xc0, + 0x7b, 0x03, 0x89, 0x4d, 0x74, 0xc7, 0xae, 0x38, 0x84, 0x8a, 0x4f, 0x6d, 0xe6, 0x53, 0x2b, 0x88, + 0x13, 0xfd, 0x66, 0x3b, 0x0d, 0x89, 0xd1, 0x09, 0x51, 0xea, 0xae, 0x9d, 0x40, 0x23, 0x17, 0x03, + 0xbf, 0x80, 0xbd, 0x34, 0x03, 0xc3, 0x72, 0x17, 0x3c, 0x3a, 0x89, 0x5d, 0xd2, 0x4c, 0xcd, 0x83, + 0xd0, 0xaa, 0x1d, 0xc3, 0xde, 0x0a, 0x6d, 0x38, 0x19, 0x9c, 0xde, 0x1a, 0x53, 0xd9, 0x28, 0x69, + 0x0d, 0xe9, 0x6d, 0xd4, 0x39, 0xfa, 0xef, 0x0a, 0xec, 0x47, 0xc1, 0x16, 0x3e, 0x9d, 0x84, 0x05, + 0x4c, 0xce, 0xfc, 0x21, 0xb5, 0x7f, 0x01, 0x0d, 0x2b, 0x26, 0x33, 0xd3, 0xfa, 0xd7, 0xfa, 0x9d, + 0x7c, 0x25, 0x64, 0x98, 0x41, 0x16, 0x47, 0xf2, 0x6e, 0x7a, 0x1b, 0x1e, 0xaf, 0x8a, 0x8a, 0xaa, + 0xa6, 0x13, 0x38, 0xf8, 0x9e, 0x06, 0x1b, 0x18, 0x1e, 0xae, 0x5a, 0x9f, 0xc1, 0xe1, 0x5b, 0x38, + 0xe3, 0x06, 0x59, 0x4b, 0x4b, 0x79, 0x58, 0x5a, 0x16, 0x3c, 0x7a, 0xc1, 0xb8, 0x1d, 0x95, 0xfe, + 0xff, 0xd4, 0x59, 0x83, 0x8a, 0x67, 0xfa, 0xd9, 0x16, 0x4f, 0xd7, 0xfa, 0xd7, 0x80, 0xd9, 0x20, + 0x71, 0x0a, 0x8f, 0xa1, 0x94, 0x6b, 0x81, 0x78, 0xa5, 0xff, 0xa1, 0x02, 0xae, 0x0b, 0x7f, 0xe7, + 0x8e, 0xc3, 0x8f, 0x01, 0x2c, 0xd7, 0x71, 0xa8, 0x25, 0xb5, 0x44, 0x22, 0x33, 0x96, 0xf0, 0x9d, + 0xf0, 0xa9, 0xe7, 0x30, 0x6b, 0xd9, 0x0f, 0x55, 0x92, 0x35, 0xe1, 0xa7, 0x50, 0x67, 0xc2, 0x08, + 0x7c, 0x93, 0x0b, 0x46, 0x79, 0x20, 0x5f, 0xaa, 0x0a, 0xa9, 0x31, 0x31, 0x49, 0x4c, 0xf8, 0x0a, + 0x6a, 0x51, 0x58, 0x97, 0x33, 0x3e, 0x93, 0x8f, 0x4d, 0x73, 0x75, 0xbc, 0xd6, 0x93, 0xe8, 0x8d, + 0x12, 0xa9, 0x8c, 0xcf, 0x48, 0x96, 0x41, 0x7f, 0x0e, 0xf5, 0xec, 0x26, 0x22, 0x34, 0xaf, 0x5c, + 0x7e, 0xb5, 0x70, 0x9c, 0x1f, 0xe8, 0xdd, 0x4b, 0x53, 0x5c, 0xb7, 0x76, 0xb0, 0x06, 0xe5, 0x64, + 0xa1, 0x60, 0x13, 0x80, 0xb8, 0x0b, 0x6e, 0x13, 0x77, 0xca, 0x78, 0x4b, 0xed, 0xff, 0x5d, 0x80, + 0xd6, 0x98, 0x9a, 0xb7, 0x94, 0xda, 0x97, 0x89, 0x0a, 0x7c, 0x05, 0xd5, 0xf4, 0x45, 0xc5, 0x4f, + 0xee, 0x79, 0x6a, 0xb5, 0x8f, 0xf2, 0x80, 0xdc, 0x4d, 0xae, 0xef, 0x74, 0x95, 0x63, 0x05, 0x2f, + 0xa0, 0x1c, 0xdf, 0x1a, 0x78, 0xb0, 0xed, 0xce, 0xd5, 0x0e, 0xb7, 0x5e, 0x35, 0x31, 0xdb, 0xcf, + 0xd0, 0xcc, 0x0f, 0x15, 0x3e, 0xc9, 0xbb, 0x6d, 0xbc, 0x07, 0xb4, 0xcf, 0xb6, 0x83, 0x92, 0x10, + 0xe8, 0xc3, 0xfe, 0xc6, 0x29, 0xc2, 0x95, 0xdf, 0xa3, 0x6d, 0xe3, 0xab, 0x1d, 0xbd, 0x13, 0x36, + 0x8d, 0xf9, 0x23, 0xc0, 0xb2, 0xd7, 0x57, 0x4b, 0xbe, 0x36, 0x6a, 0x5a, 0xe7, 0xed, 0x80, 0x84, + 0xf2, 0x4c, 0x87, 0x96, 0x88, 0x0e, 0xf6, 0xb5, 0xe8, 0x59, 0x4e, 0xd8, 0x7f, 0x67, 0xcd, 0xf4, + 0x8c, 0x47, 0xe1, 0x2f, 0xe6, 0xb4, 0x24, 0xff, 0x34, 0xbf, 0xfd, 0x37, 0x00, 0x00, 0xff, 0xff, + 0xad, 0x6b, 0x26, 0x8c, 0x7c, 0x0a, 0x00, 0x00, } diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 8b02d2bba..ca96b8d20 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -1051,12 +1051,10 @@ func (m *VolumeEcShardsGenerateRequest) GetCollection() string { type VolumeEcShardsGenerateResponse struct { } -func (m *VolumeEcShardsGenerateResponse) Reset() { *m = VolumeEcShardsGenerateResponse{} } -func (m *VolumeEcShardsGenerateResponse) String() string { return proto.CompactTextString(m) } -func (*VolumeEcShardsGenerateResponse) ProtoMessage() {} -func (*VolumeEcShardsGenerateResponse) Descriptor() ([]byte, []int) { - return fileDescriptor0, []int{41} -} +func (m *VolumeEcShardsGenerateResponse) Reset() { *m = VolumeEcShardsGenerateResponse{} } +func (m *VolumeEcShardsGenerateResponse) String() string { return proto.CompactTextString(m) } +func (*VolumeEcShardsGenerateResponse) ProtoMessage() {} +func (*VolumeEcShardsGenerateResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{41} } type VolumeEcShardsRebuildRequest struct { VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,omitempty"` @@ -1429,12 +1427,10 @@ func (m *VolumeEcShardsToVolumeRequest) GetCollection() string { type VolumeEcShardsToVolumeResponse struct { } -func (m *VolumeEcShardsToVolumeResponse) Reset() { *m = VolumeEcShardsToVolumeResponse{} } -func (m *VolumeEcShardsToVolumeResponse) String() string { return proto.CompactTextString(m) } -func (*VolumeEcShardsToVolumeResponse) ProtoMessage() {} -func (*VolumeEcShardsToVolumeResponse) Descriptor() ([]byte, []int) { - return fileDescriptor0, []int{57} -} +func (m *VolumeEcShardsToVolumeResponse) Reset() { *m = VolumeEcShardsToVolumeResponse{} } +func (m *VolumeEcShardsToVolumeResponse) String() string { return proto.CompactTextString(m) } +func (*VolumeEcShardsToVolumeResponse) ProtoMessage() {} +func (*VolumeEcShardsToVolumeResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{57} } type ReadVolumeFileStatusRequest struct { VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,omitempty"` @@ -2101,10 +2097,8 @@ type QueryRequest_InputSerialization_JSONInput struct { func (m *QueryRequest_InputSerialization_JSONInput) Reset() { *m = QueryRequest_InputSerialization_JSONInput{} } -func (m *QueryRequest_InputSerialization_JSONInput) String() string { - return proto.CompactTextString(m) -} -func (*QueryRequest_InputSerialization_JSONInput) ProtoMessage() {} +func (m *QueryRequest_InputSerialization_JSONInput) String() string { return proto.CompactTextString(m) } +func (*QueryRequest_InputSerialization_JSONInput) ProtoMessage() {} func (*QueryRequest_InputSerialization_JSONInput) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{70, 1, 1} } diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index a790e4601..266030f5d 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -381,3 +381,39 @@ func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb. return t, nil } + +func (fs *FilerServer) KeepConnected(stream filer_pb.SeaweedFiler_KeepConnectedServer) error { + + req, err := stream.Recv() + if err != nil { + return err + } + + clientName := fmt.Sprintf("%s:%d", req.Name, req.GrpcPort) + fs.brokersLock.Lock() + fs.brokers[clientName] = true + glog.V(0).Infof("+ broker %v", clientName) + fs.brokersLock.Unlock() + + defer func() { + fs.brokersLock.Lock() + delete(fs.brokers, clientName) + glog.V(0).Infof("- broker %v: %v", clientName, err) + fs.brokersLock.Unlock() + }() + + for { + if err := stream.Send(&filer_pb.KeepConnectedResponse{}); err != nil { + glog.V(0).Infof("send broker %v: %+v", clientName, err) + return err + } + // println("replied") + + if _, err := stream.Recv(); err != nil { + glog.V(0).Infof("recv broker %v: %v", clientName, err) + return err + } + // println("received") + } + +} diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 956684d46..0a54ea97d 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -8,9 +8,10 @@ import ( "sync" "time" - "github.com/chrislusf/seaweedfs/weed/util/grace" "google.golang.org/grpc" + "github.com/chrislusf/seaweedfs/weed/util/grace" + "github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb" @@ -62,6 +63,9 @@ type FilerServer struct { // notifying clients listenersLock sync.Mutex listenersCond *sync.Cond + + brokers map[string]bool + brokersLock sync.Mutex } func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) { @@ -69,6 +73,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) fs = &FilerServer{ option: option, grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"), + brokers: make(map[string]bool), } fs.listenersCond = sync.NewCond(&fs.listenersLock)