gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

publish and send to follower

Browse Source
This commit is contained in:
chrislu
2024-03-27 23:10:24 -07:00
parent f249f831b1
commit dab5454332
2 changed files with 23 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -143,21 +143,8 @@ func (b *MessageQueueBroker) PublishMessage(stream mq_pb.SeaweedMessaging_Publis
} }
// send to the local partition // send to the local partition
localTopicPartition.Publish(dataMessage) if err = localTopicPartition.Publish(dataMessage); err != nil {
receivedSequence = dataMessage.TsNs return fmt.Errorf("topic %v partition %v publish error: %v", initMessage.Topic, initMessage.Partition, err)
// maybe send to the follower
if localTopicPartition.FollowerStream != nil {
println("recv", string(dataMessage.Key), dataMessage.TsNs)
if followErr := localTopicPartition.FollowerStream.Send(&mq_pb.PublishFollowMeRequest{
Message: &mq_pb.PublishFollowMeRequest_Data{
Data: dataMessage,
},
}); followErr != nil {
return followErr
}
} else {
atomic.StoreInt64(&localTopicPartition.AckTsNs, receivedSequence)
} }
} }

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

@@ -32,6 +32,7 @@ type LocalPartition struct {
FollowerStream mq_pb.SeaweedMessaging_PublishFollowMeClient FollowerStream mq_pb.SeaweedMessaging_PublishFollowMeClient
FollowerGrpcConnection *grpc.ClientConn FollowerGrpcConnection *grpc.ClientConn
follower string
} }
var TIME_FORMAT = "2006-01-02-15-04-05" var TIME_FORMAT = "2006-01-02-15-04-05"
@@ -54,8 +55,24 @@ func NewLocalPartition(partition Partition, isLeader bool, followerBrokers []pb.
return lp return lp
} }
func (p *LocalPartition) Publish(message *mq_pb.DataMessage) { func (p *LocalPartition) Publish(message *mq_pb.DataMessage) error {
p.LogBuffer.AddToBuffer(message.Key, message.Value, time.Now().UnixNano()) p.LogBuffer.AddToBuffer(message.Key, message.Value, time.Now().UnixNano())
// maybe send to the follower
if p.FollowerStream != nil {
println("recv", string(message.Key), message.TsNs)
if followErr := p.FollowerStream.Send(&mq_pb.PublishFollowMeRequest{
Message: &mq_pb.PublishFollowMeRequest_Data{
Data: message,
},
}); followErr != nil {
return fmt.Errorf("send to follower %s: %v", p.follower, followErr)
}
} else {
atomic.StoreInt64(&p.AckTsNs, message.TsNs)
}
return nil
} }
func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.MessagePosition, func (p *LocalPartition) Subscribe(clientName string, startPosition log_buffer.MessagePosition,
@@ -137,11 +154,11 @@ func (p *LocalPartition) MaybeConnectToFollowers(initMessage *mq_pb.PublishMessa
return nil return nil
} }
follower := initMessage.FollowerBrokers[0] p.follower = initMessage.FollowerBrokers[0]
ctx := context.Background() ctx := context.Background()
p.FollowerGrpcConnection, err = pb.GrpcDial(ctx, follower, true, grpcDialOption) p.FollowerGrpcConnection, err = pb.GrpcDial(ctx, p.follower, true, grpcDialOption)
if err != nil { if err != nil {
return fmt.Errorf("fail to dial %s: %v", follower, err) return fmt.Errorf("fail to dial %s: %v", p.follower, err)
} }
followerClient := mq_pb.NewSeaweedMessagingClient(p.FollowerGrpcConnection) followerClient := mq_pb.NewSeaweedMessagingClient(p.FollowerGrpcConnection)
p.FollowerStream, err = followerClient.PublishFollowMe(ctx) p.FollowerStream, err = followerClient.PublishFollowMe(ctx)