sub

weed/mq/broker/brokder_grpc_sub.go

@@ -0,0 +1,40 @@
+package broker
+
+import (
+	"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 {
+
+	localTopicPartition := broker.localTopicManager.GetTopicPartition(topic.FromPbTopic(req.Init.Topic),
+		topic.FromPbPartition(req.Init.Partition))
+	if localTopicPartition == nil {
+		stream.Send(&mq_pb.SubscribeResponse{
+			Message: &mq_pb.SubscribeResponse_Ctrl{
+				&mq_pb.SubscribeResponse_CtrlMessage{
+					Error: "not found",
+				},
+			},
+		})
+		return nil
+	}
+
+	localTopicPartition.Subscribe("client", time.Now(), func(logEntry *filer_pb.LogEntry) error {
+		if err := stream.Send(&mq_pb.SubscribeResponse{Message: &mq_pb.SubscribeResponse_Data{
+			Data: &mq_pb.DataMessage{
+				Key:   []byte("hello"),
+				Value: []byte("world"),
+			},
+		}}); err != nil {
+			glog.Errorf("Error sending setup response: %v", err)
+			return err
+		}
+		return nil
+	})
+
+	return nil
+}

weed/mq/client/pub_client/publisher.go

@@ -0,0 +1,5 @@
+package main
+
+func main() {
+
+}

weed/mq/topic/local_partition.go

@@ -2,6 +2,7 @@ package topic
 
 import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 	"time"
@@ -14,10 +15,18 @@ type LocalPartition struct {
 	logBuffer       *log_buffer.LogBuffer
 }
 
+type OnEachMessageFn func(logEntry *filer_pb.LogEntry) error
+
 func (p LocalPartition) Publish(message *mq_pb.DataMessage) {
 	p.logBuffer.AddToBuffer(message.Key, message.Value, time.Now().UnixNano())
 }
 
+func (p LocalPartition) Subscribe(clientName string, startReadTime time.Time, eachMessageFn OnEachMessageFn) {
+	p.logBuffer.LoopProcessLogData(clientName, startReadTime, 0, func() bool {
+		return true
+	}, eachMessageFn)
+}
+
 func FromPbBrokerPartitionsAssignment(self pb.ServerAddress, assignment *mq_pb.BrokerPartitionsAssignment) *LocalPartition {
 	isLeaer := assignment.LeaderBroker == string(self)
 	localPartition := &LocalPartition{