mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-09-18 23:10:35 +02:00
Code Issues Projects Releases Wiki Activity

rename functions

Browse source
This commit is contained in:
chrislu 2024-01-05 15:14:25 -08:00
parent 1fed37bed0
commit ee41dbb7fc
7 changed files with 434 additions and 426 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
weed/mq/broker/broker_grpc_pub.go
View file

@ -34,7 +34,7 @@ import (
// Subscribers needs to listen for new partitions and connect to the brokers.
// Each subscription may not get data. It can act as a backup.
func (b *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer) error {
func (b *MessageQueueBroker) PublishMessage(stream mq_pb.SeaweedMessaging_PublishMessageServer) error {
// 1. write to the volume server
// 2. find the topic metadata owning filer
// 3. write to the filer
@ -44,7 +44,7 @@ func (b *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer
if err != nil {
return err
}
response := &mq_pb.PublishResponse{}
response := &mq_pb.PublishMessageResponse{}
// TODO check whether current broker should be the leader for the topic partition
ackInterval := 1
initMessage := req.GetInit()
@ -70,7 +70,7 @@ func (b *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer
ackCounter := 0
var ackSequence int64
var isStopping int32
respChan := make(chan *mq_pb.PublishResponse, 128)
respChan := make(chan *mq_pb.PublishMessageResponse, 128)
defer func() {
atomic.StoreInt32(&isStopping, 1)
close(respChan)
@ -90,7 +90,7 @@ func (b *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer
}
case <-ticker.C:
if atomic.LoadInt32(&isStopping) == 0 {
response := &mq_pb.PublishResponse{
response := &mq_pb.PublishMessageResponse{
AckSequence: ackSequence,
}
respChan <- response
@ -98,7 +98,7 @@ func (b *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer
return
}
case <-localTopicPartition.StopPublishersCh:
respChan <- &mq_pb.PublishResponse{
respChan <- &mq_pb.PublishMessageResponse{
AckSequence: ackSequence,
ShouldClose: true,
}
@ -124,7 +124,7 @@ func (b *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer
if ackCounter >= ackInterval {
ackCounter = 0
// send back the ack
response := &mq_pb.PublishResponse{
response := &mq_pb.PublishMessageResponse{
AckSequence: ackSequence,
}
respChan <- response

10
weed/mq/broker/broker_grpc_sub.go
View file

@ -10,15 +10,15 @@ import (
"time"
)
func (b *MessageQueueBroker) Subscribe(req *mq_pb.SubscribeRequest, stream mq_pb.SeaweedMessaging_SubscribeServer) error {
func (b *MessageQueueBroker) SubscribeMessage(req *mq_pb.SubscribeMessageRequest, stream mq_pb.SeaweedMessaging_SubscribeMessageServer) error {
t := topic.FromPbTopic(req.GetInit().Topic)
partition := topic.FromPbPartition(req.GetInit().Partition)
localTopicPartition := b.localTopicManager.GetTopicPartition(t, partition)
if localTopicPartition == nil {
stream.Send(&mq_pb.SubscribeResponse{
Message: &mq_pb.SubscribeResponse_Ctrl{
Ctrl: &mq_pb.SubscribeResponse_CtrlMessage{
stream.Send(&mq_pb.SubscribeMessageResponse{
Message: &mq_pb.SubscribeMessageResponse_Ctrl{
Ctrl: &mq_pb.SubscribeMessageResponse_CtrlMessage{
Error: "not initialized",
},
},
@ -73,7 +73,7 @@ func (b *MessageQueueBroker) Subscribe(req *mq_pb.SubscribeRequest, stream mq_pb
sleepIntervalCount = 0
value := logEntry.GetData()
if err := stream.Send(&mq_pb.SubscribeResponse{Message: &mq_pb.SubscribeResponse_Data{
if err := stream.Send(&mq_pb.SubscribeMessageResponse{Message: &mq_pb.SubscribeMessageResponse_Data{
Data: &mq_pb.DataMessage{
Key: []byte(fmt.Sprintf("key-%d", logEntry.PartitionKeyHash)),
Value: value,

8
weed/mq/client/pub_client/connect.go
View file

@ -21,7 +21,7 @@ func (p *TopicPublisher) doConnect(partition *mq_pb.Partition, brokerAddress str
return publishClient, fmt.Errorf("dial broker %s: %v", brokerAddress, err)
}
brokerClient := mq_pb.NewSeaweedMessagingClient(grpcConnection)
stream, err := brokerClient.Publish(context.Background())
stream, err := brokerClient.SubscribeMessage(context.Background())
if err != nil {
return publishClient, fmt.Errorf("create publish client: %v", err)
}
@ -29,9 +29,9 @@ func (p *TopicPublisher) doConnect(partition *mq_pb.Partition, brokerAddress str
SeaweedMessaging_PublishClient: stream,
Broker: brokerAddress,
}
if err = publishClient.Send(&mq_pb.PublishRequest{
Message: &mq_pb.PublishRequest_Init{
Init: &mq_pb.PublishRequest_InitMessage{
if err = publishClient.Send(&mq_pb.PublishMessageRequest{
Message: &mq_pb.PublishMessageRequest_Init{
Init: &mq_pb.PublishMessageRequest_InitMessage{
Topic: &mq_pb.Topic{
Namespace: p.namespace,
Name: p.topic,

4
weed/mq/client/pub_client/publish.go
View file

@ -27,8 +27,8 @@ func (p *TopicPublisher) Publish(key, value []byte) error {
//google.golang.org/grpc.(*clientStream).SendMsg(stream.go:894)
//github.com/seaweedfs/seaweedfs/weed/pb/mq_pb.(*seaweedMessagingPublishClient).Send(mq_grpc.pb.go:141)
//github.com/seaweedfs/seaweedfs/weed/mq/client/pub_client.(*TopicPublisher).Publish(publish.go:19)
if err := publishClient.Send(&mq_pb.PublishRequest{
Message: &mq_pb.PublishRequest_Data{
if err := publishClient.Send(&mq_pb.PublishMessageRequest{
Message: &mq_pb.PublishMessageRequest_Data{
Data: &mq_pb.DataMessage{
Key: key,
Value: value,

12
weed/pb/mq.proto
View file

@ -41,9 +41,9 @@ service SeaweedMessaging {
}
// data plane for each topic partition
rpc Publish (stream PublishRequest) returns (stream PublishResponse) {
rpc PublishMessage (stream PublishMessageRequest) returns (stream PublishMessageResponse) {
}
rpc Subscribe (SubscribeRequest) returns (stream SubscribeResponse) {
rpc SubscribeMessage (SubscribeMessageRequest) returns (stream SubscribeMessageResponse) {
}
}
@ -180,7 +180,7 @@ message DataMessage {
bytes value = 2;
int64 ts_ns = 3;
}
message PublishRequest {
message PublishMessageRequest {
message InitMessage {
Topic topic = 1;
Partition partition = 2;
@ -192,12 +192,12 @@ message PublishRequest {
}
int64 sequence = 3;
}
message PublishResponse {
message PublishMessageResponse {
int64 ack_sequence = 1;
string error = 2;
bool should_close = 3;
}
message SubscribeRequest {
message SubscribeMessageRequest {
message InitMessage {
string consumer_group = 1;
string consumer_id = 2;
@ -218,7 +218,7 @@ message SubscribeRequest {
AckMessage ack = 2;
}
}
message SubscribeResponse {
message SubscribeMessageResponse {
message CtrlMessage {
string error = 1;
bool is_end_of_stream = 2;

718
weed/pb/mq_pb/mq.pb.go
View file

File diff suppressed because it is too large Load diff

96
weed/pb/mq_pb/mq_grpc.pb.go
View file

@ -29,8 +29,8 @@ const (
SeaweedMessaging_ClosePublishers_FullMethodName = "/messaging_pb.SeaweedMessaging/ClosePublishers"
SeaweedMessaging_CloseSubscribers_FullMethodName = "/messaging_pb.SeaweedMessaging/CloseSubscribers"
SeaweedMessaging_SubscriberToSubCoordinator_FullMethodName = "/messaging_pb.SeaweedMessaging/SubscriberToSubCoordinator"
SeaweedMessaging_Publish_FullMethodName = "/messaging_pb.SeaweedMessaging/Publish"
SeaweedMessaging_Subscribe_FullMethodName = "/messaging_pb.SeaweedMessaging/Subscribe"
SeaweedMessaging_PublishMessage_FullMethodName = "/messaging_pb.SeaweedMessaging/PublishMessage"
SeaweedMessaging_SubscribeMessage_FullMethodName = "/messaging_pb.SeaweedMessaging/SubscribeMessage"
)
// SeaweedMessagingClient is the client API for SeaweedMessaging service.
@ -53,8 +53,8 @@ type SeaweedMessagingClient interface {
// subscriber connects to broker balancer, which coordinates with the subscribers
SubscriberToSubCoordinator(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_SubscriberToSubCoordinatorClient, error)
// data plane for each topic partition
Publish(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishClient, error)
Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (SeaweedMessaging_SubscribeClient, error)
PublishMessage(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishMessageClient, error)
SubscribeMessage(ctx context.Context, in *SubscribeMessageRequest, opts ...grpc.CallOption) (SeaweedMessaging_SubscribeMessageClient, error)
}
type seaweedMessagingClient struct {
@ -199,43 +199,43 @@ func (x *seaweedMessagingSubscriberToSubCoordinatorClient) Recv() (*SubscriberTo
return m, nil
}
func (c *seaweedMessagingClient) Publish(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishClient, error) {
stream, err := c.cc.NewStream(ctx, &SeaweedMessaging_ServiceDesc.Streams[2], SeaweedMessaging_Publish_FullMethodName, opts...)
func (c *seaweedMessagingClient) PublishMessage(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishMessageClient, error) {
stream, err := c.cc.NewStream(ctx, &SeaweedMessaging_ServiceDesc.Streams[2], SeaweedMessaging_PublishMessage_FullMethodName, opts...)
if err != nil {
return nil, err
}
x := &seaweedMessagingPublishClient{stream}
x := &seaweedMessagingPublishMessageClient{stream}
return x, nil
}
type SeaweedMessaging_PublishClient interface {
Send(*PublishRequest) error
Recv() (*PublishResponse, error)
type SeaweedMessaging_PublishMessageClient interface {
Send(*PublishMessageRequest) error
Recv() (*PublishMessageResponse, error)
grpc.ClientStream
}
type seaweedMessagingPublishClient struct {
type seaweedMessagingPublishMessageClient struct {
grpc.ClientStream
}
func (x *seaweedMessagingPublishClient) Send(m *PublishRequest) error {
func (x *seaweedMessagingPublishMessageClient) Send(m *PublishMessageRequest) error {
return x.ClientStream.SendMsg(m)
}
func (x *seaweedMessagingPublishClient) Recv() (*PublishResponse, error) {
m := new(PublishResponse)
func (x *seaweedMessagingPublishMessageClient) Recv() (*PublishMessageResponse, error) {
m := new(PublishMessageResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func (c *seaweedMessagingClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (SeaweedMessaging_SubscribeClient, error) {
stream, err := c.cc.NewStream(ctx, &SeaweedMessaging_ServiceDesc.Streams[3], SeaweedMessaging_Subscribe_FullMethodName, opts...)
func (c *seaweedMessagingClient) SubscribeMessage(ctx context.Context, in *SubscribeMessageRequest, opts ...grpc.CallOption) (SeaweedMessaging_SubscribeMessageClient, error) {
stream, err := c.cc.NewStream(ctx, &SeaweedMessaging_ServiceDesc.Streams[3], SeaweedMessaging_SubscribeMessage_FullMethodName, opts...)
if err != nil {
return nil, err
}
x := &seaweedMessagingSubscribeClient{stream}
x := &seaweedMessagingSubscribeMessageClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
@ -245,17 +245,17 @@ func (c *seaweedMessagingClient) Subscribe(ctx context.Context, in *SubscribeReq
return x, nil
}
type SeaweedMessaging_SubscribeClient interface {
Recv() (*SubscribeResponse, error)
type SeaweedMessaging_SubscribeMessageClient interface {
Recv() (*SubscribeMessageResponse, error)
grpc.ClientStream
}
type seaweedMessagingSubscribeClient struct {
type seaweedMessagingSubscribeMessageClient struct {
grpc.ClientStream
}
func (x *seaweedMessagingSubscribeClient) Recv() (*SubscribeResponse, error) {
m := new(SubscribeResponse)
func (x *seaweedMessagingSubscribeMessageClient) Recv() (*SubscribeMessageResponse, error) {
m := new(SubscribeMessageResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
@ -282,8 +282,8 @@ type SeaweedMessagingServer interface {
// subscriber connects to broker balancer, which coordinates with the subscribers
SubscriberToSubCoordinator(SeaweedMessaging_SubscriberToSubCoordinatorServer) error
// data plane for each topic partition
Publish(SeaweedMessaging_PublishServer) error
Subscribe(*SubscribeRequest, SeaweedMessaging_SubscribeServer) error
PublishMessage(SeaweedMessaging_PublishMessageServer) error
SubscribeMessage(*SubscribeMessageRequest, SeaweedMessaging_SubscribeMessageServer) error
mustEmbedUnimplementedSeaweedMessagingServer()
}
@ -321,11 +321,11 @@ func (UnimplementedSeaweedMessagingServer) CloseSubscribers(context.Context, *Cl
func (UnimplementedSeaweedMessagingServer) SubscriberToSubCoordinator(SeaweedMessaging_SubscriberToSubCoordinatorServer) error {
return status.Errorf(codes.Unimplemented, "method SubscriberToSubCoordinator not implemented")
}
func (UnimplementedSeaweedMessagingServer) Publish(SeaweedMessaging_PublishServer) error {
return status.Errorf(codes.Unimplemented, "method Publish not implemented")
func (UnimplementedSeaweedMessagingServer) PublishMessage(SeaweedMessaging_PublishMessageServer) error {
return status.Errorf(codes.Unimplemented, "method PublishMessage not implemented")
}
func (UnimplementedSeaweedMessagingServer) Subscribe(*SubscribeRequest, SeaweedMessaging_SubscribeServer) error {
return status.Errorf(codes.Unimplemented, "method Subscribe not implemented")
func (UnimplementedSeaweedMessagingServer) SubscribeMessage(*SubscribeMessageRequest, SeaweedMessaging_SubscribeMessageServer) error {
return status.Errorf(codes.Unimplemented, "method SubscribeMessage not implemented")
}
func (UnimplementedSeaweedMessagingServer) mustEmbedUnimplementedSeaweedMessagingServer() {}
@ -536,50 +536,50 @@ func (x *seaweedMessagingSubscriberToSubCoordinatorServer) Recv() (*SubscriberTo
return m, nil
}
func _SeaweedMessaging_Publish_Handler(srv interface{}, stream grpc.ServerStream) error {
return srv.(SeaweedMessagingServer).Publish(&seaweedMessagingPublishServer{stream})
func _SeaweedMessaging_PublishMessage_Handler(srv interface{}, stream grpc.ServerStream) error {
return srv.(SeaweedMessagingServer).PublishMessage(&seaweedMessagingPublishMessageServer{stream})
}
type SeaweedMessaging_PublishServer interface {
Send(*PublishResponse) error
Recv() (*PublishRequest, error)
type SeaweedMessaging_PublishMessageServer interface {
Send(*PublishMessageResponse) error
Recv() (*PublishMessageRequest, error)
grpc.ServerStream
}
type seaweedMessagingPublishServer struct {
type seaweedMessagingPublishMessageServer struct {
grpc.ServerStream
}
func (x *seaweedMessagingPublishServer) Send(m *PublishResponse) error {
func (x *seaweedMessagingPublishMessageServer) Send(m *PublishMessageResponse) error {
return x.ServerStream.SendMsg(m)
}
func (x *seaweedMessagingPublishServer) Recv() (*PublishRequest, error) {
m := new(PublishRequest)
func (x *seaweedMessagingPublishMessageServer) Recv() (*PublishMessageRequest, error) {
m := new(PublishMessageRequest)
if err := x.ServerStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func _SeaweedMessaging_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeRequest)
func _SeaweedMessaging_SubscribeMessage_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeMessageRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(SeaweedMessagingServer).Subscribe(m, &seaweedMessagingSubscribeServer{stream})
return srv.(SeaweedMessagingServer).SubscribeMessage(m, &seaweedMessagingSubscribeMessageServer{stream})
}
type SeaweedMessaging_SubscribeServer interface {
Send(*SubscribeResponse) error
type SeaweedMessaging_SubscribeMessageServer interface {
Send(*SubscribeMessageResponse) error
grpc.ServerStream
}
type seaweedMessagingSubscribeServer struct {
type seaweedMessagingSubscribeMessageServer struct {
grpc.ServerStream
}
func (x *seaweedMessagingSubscribeServer) Send(m *SubscribeResponse) error {
func (x *seaweedMessagingSubscribeMessageServer) Send(m *SubscribeMessageResponse) error {
return x.ServerStream.SendMsg(m)
}
@ -637,14 +637,14 @@ var SeaweedMessaging_ServiceDesc = grpc.ServiceDesc{
ClientStreams: true,
},
{
StreamName: "Publish",
Handler: _SeaweedMessaging_Publish_Handler,
StreamName: "PublishMessage",
Handler: _SeaweedMessaging_PublishMessage_Handler,
ServerStreams: true,
ClientStreams: true,
},
{
StreamName: "Subscribe",
Handler: _SeaweedMessaging_Subscribe_Handler,
StreamName: "SubscribeMessage",
Handler: _SeaweedMessaging_SubscribeMessage_Handler,
ServerStreams: true,
},
},