seaweedfs/weed/mq/broker/broker_grpc_sub.go

45 lines
1.3 KiB
Go
Raw Normal View History

2023-08-27 20:13:14 +00:00
package broker
import (
2023-08-28 00:50:59 +00:00
"fmt"
2023-08-27 20:13:14 +00:00
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"time"
)
func (broker *MessageQueueBroker) Subscribe(req *mq_pb.SubscribeRequest, stream mq_pb.SeaweedMessaging_SubscribeServer) error {
2023-10-01 04:26:09 +00:00
localTopicPartition := broker.localTopicManager.GetTopicPartition(topic.FromPbTopic(req.GetConsumer().Topic),
topic.FromPbPartition(req.GetConsumer().Partition))
2023-08-27 20:13:14 +00:00
if localTopicPartition == nil {
stream.Send(&mq_pb.SubscribeResponse{
Message: &mq_pb.SubscribeResponse_Ctrl{
2023-08-28 00:50:59 +00:00
Ctrl: &mq_pb.SubscribeResponse_CtrlMessage{
2023-09-01 07:36:51 +00:00
Error: "not initialized",
2023-08-27 20:13:14 +00:00
},
},
})
return nil
}
2023-10-01 04:26:09 +00:00
clientName := fmt.Sprintf("%s/%s-%s", req.GetConsumer().ConsumerGroup, req.GetConsumer().ConsumerId, req.GetConsumer().ClientId)
2023-09-01 07:36:51 +00:00
localTopicPartition.Subscribe(clientName, time.Now(), func(logEntry *filer_pb.LogEntry) error {
2023-08-28 00:50:59 +00:00
value := logEntry.GetData()
2023-08-27 20:13:14 +00:00
if err := stream.Send(&mq_pb.SubscribeResponse{Message: &mq_pb.SubscribeResponse_Data{
Data: &mq_pb.DataMessage{
2023-08-28 00:50:59 +00:00
Key: []byte(fmt.Sprintf("key-%d", logEntry.PartitionKeyHash)),
Value: value,
2023-08-27 20:13:14 +00:00
},
}}); err != nil {
glog.Errorf("Error sending setup response: %v", err)
return err
}
return nil
})
return nil
}