lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-05-08 00:07:48 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

rename

Browse Source
This commit is contained in:
chrislu 2024-03-26 21:52:12 -07:00
parent 7d69552974
commit f67c8309d8
2 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
weed/mq/broker/broker_grpc_pub.go
View File

@ -68,13 +68,13 @@ func (b *MessageQueueBroker) PublishMessage(stream mq_pb.SeaweedMessaging_Publis
if localTopicPartition.FollowerStream == nil && len(initMessage.FollowerBrokers) > 0 { if localTopicPartition.FollowerStream == nil && len(initMessage.FollowerBrokers) > 0 {
follower := initMessage.FollowerBrokers[0] follower := initMessage.FollowerBrokers[0]
ctx := stream.Context() ctx := stream.Context()
localTopicPartition.GrpcConnection, err = pb.GrpcDial(ctx, follower, true, b.grpcDialOption) localTopicPartition.FollowerGrpcConnection, err = pb.GrpcDial(ctx, follower, true, b.grpcDialOption)
if err != nil { if err != nil {
response.Error = fmt.Sprintf("fail to dial %s: %v", follower, err) response.Error = fmt.Sprintf("fail to dial %s: %v", follower, err)
glog.Errorf("fail to dial %s: %v", follower, err) glog.Errorf("fail to dial %s: %v", follower, err)
return stream.Send(response) return stream.Send(response)
} }
followerClient := mq_pb.NewSeaweedMessagingClient(localTopicPartition.GrpcConnection) followerClient := mq_pb.NewSeaweedMessagingClient(localTopicPartition.FollowerGrpcConnection)
localTopicPartition.FollowerStream, err = followerClient.PublishFollowMe(ctx) localTopicPartition.FollowerStream, err = followerClient.PublishFollowMe(ctx)
if err != nil { if err != nil {
response.Error = fmt.Sprintf("fail to create publish client: %v", err) response.Error = fmt.Sprintf("fail to create publish client: %v", err)
@ -136,7 +136,7 @@ func (b *MessageQueueBroker) PublishMessage(stream mq_pb.SeaweedMessaging_Publis
glog.Errorf("Error closing follower stream: %v", followErr) glog.Errorf("Error closing follower stream: %v", followErr)
} }
println("closing grpcConnection to follower") println("closing grpcConnection to follower")
localTopicPartition.GrpcConnection.Close() localTopicPartition.FollowerGrpcConnection.Close()
} }
b.localTopicManager.RemoveTopicPartition(t, p) b.localTopicManager.RemoveTopicPartition(t, p)
glog.V(0).Infof("Removed local topic %v partition %v", initMessage.Topic, initMessage.Partition) glog.V(0).Infof("Removed local topic %v partition %v", initMessage.Topic, initMessage.Partition)

4
weed/mq/topic/local_partition.go
View File

@ -28,8 +28,8 @@ type LocalPartition struct {
Subscribers *LocalPartitionSubscribers Subscribers *LocalPartitionSubscribers
FollowerId int32 FollowerId int32
FollowerStream mq_pb.SeaweedMessaging_PublishFollowMeClient FollowerStream mq_pb.SeaweedMessaging_PublishFollowMeClient
GrpcConnection *grpc.ClientConn FollowerGrpcConnection *grpc.ClientConn
} }
var TIME_FORMAT = "2006-01-02-15-04-05" var TIME_FORMAT = "2006-01-02-15-04-05"