From eb2d2979b16308ae93e435e430e2c910cd164830 Mon Sep 17 00:00:00 2001 From: chrislu Date: Tue, 17 Sep 2024 23:11:23 -0700 Subject: [PATCH] rename --- weed/mq/topic/local_partition.go | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/weed/mq/topic/local_partition.go b/weed/mq/topic/local_partition.go index 8911c1841..21198d580 100644 --- a/weed/mq/topic/local_partition.go +++ b/weed/mq/topic/local_partition.go @@ -28,7 +28,7 @@ type LocalPartition struct { Publishers *LocalPartitionPublishers Subscribers *LocalPartitionSubscribers - publishFolloweMeStream mq_pb.SeaweedMessaging_PublishFollowMeClient + publishFollowMeStream mq_pb.SeaweedMessaging_PublishFollowMeClient followerGrpcConnection *grpc.ClientConn Follower string } @@ -55,9 +55,9 @@ func (p *LocalPartition) Publish(message *mq_pb.DataMessage) error { p.LogBuffer.AddToBuffer(message) // maybe send to the follower - if p.publishFolloweMeStream != nil { + if p.publishFollowMeStream != nil { // println("recv", string(message.Key), message.TsNs) - if followErr := p.publishFolloweMeStream.Send(&mq_pb.PublishFollowMeRequest{ + if followErr := p.publishFollowMeStream.Send(&mq_pb.PublishFollowMeRequest{ Message: &mq_pb.PublishFollowMeRequest_Data{ Data: message, }, @@ -134,7 +134,7 @@ func (p *LocalPartition) WaitUntilNoPublishers() { } func (p *LocalPartition) MaybeConnectToFollowers(initMessage *mq_pb.PublishMessageRequest_InitMessage, grpcDialOption grpc.DialOption) (err error) { - if p.publishFolloweMeStream != nil { + if p.publishFollowMeStream != nil { return nil } if initMessage.FollowerBroker == "" { @@ -148,11 +148,11 @@ func (p *LocalPartition) MaybeConnectToFollowers(initMessage *mq_pb.PublishMessa return fmt.Errorf("fail to dial %s: %v", p.Follower, err) } followerClient := mq_pb.NewSeaweedMessagingClient(p.followerGrpcConnection) - p.publishFolloweMeStream, err = followerClient.PublishFollowMe(ctx) + p.publishFollowMeStream, err = followerClient.PublishFollowMe(ctx) if err != nil { return fmt.Errorf("fail to create publish client: %v", err) } - if err = p.publishFolloweMeStream.Send(&mq_pb.PublishFollowMeRequest{ + if err = p.publishFollowMeStream.Send(&mq_pb.PublishFollowMeRequest{ Message: &mq_pb.PublishFollowMeRequest_Init{ Init: &mq_pb.PublishFollowMeRequest_InitMessage{ Topic: initMessage.Topic, @@ -170,7 +170,7 @@ func (p *LocalPartition) MaybeConnectToFollowers(initMessage *mq_pb.PublishMessa }() for { - ack, err := p.publishFolloweMeStream.Recv() + ack, err := p.publishFollowMeStream.Recv() if err != nil { e, _ := status.FromError(err) if e.Code() == codes.Canceled { @@ -194,9 +194,9 @@ func (p *LocalPartition) MaybeShutdownLocalPartition() (hasShutdown bool) { for !p.LogBuffer.IsAllFlushed() { time.Sleep(113 * time.Millisecond) } - if p.publishFolloweMeStream != nil { + if p.publishFollowMeStream != nil { // send close to the follower - if followErr := p.publishFolloweMeStream.Send(&mq_pb.PublishFollowMeRequest{ + if followErr := p.publishFollowMeStream.Send(&mq_pb.PublishFollowMeRequest{ Message: &mq_pb.PublishFollowMeRequest_Close{ Close: &mq_pb.PublishFollowMeRequest_CloseMessage{}, }, @@ -205,7 +205,7 @@ func (p *LocalPartition) MaybeShutdownLocalPartition() (hasShutdown bool) { } glog.V(4).Infof("closing grpcConnection to follower") p.followerGrpcConnection.Close() - p.publishFolloweMeStream = nil + p.publishFollowMeStream = nil p.Follower = "" } @@ -224,8 +224,8 @@ func (p *LocalPartition) Shutdown() { } func (p *LocalPartition) NotifyLogFlushed(flushTsNs int64) { - if p.publishFolloweMeStream != nil { - if followErr := p.publishFolloweMeStream.Send(&mq_pb.PublishFollowMeRequest{ + if p.publishFollowMeStream != nil { + if followErr := p.publishFollowMeStream.Send(&mq_pb.PublishFollowMeRequest{ Message: &mq_pb.PublishFollowMeRequest_Flush{ Flush: &mq_pb.PublishFollowMeRequest_FlushMessage{ TsNs: flushTsNs,