seaweedfs/weed/messaging/broker/broker_grpc_server_publish.go

91 lines
1.7 KiB
Go
Raw Normal View History

2020-04-18 08:12:01 +00:00
package broker
2020-04-18 05:39:21 +00:00
import (
"io"
"github.com/golang/protobuf/proto"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
)
func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_PublishServer) error {
// process initial request
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
// TODO look it up
topicConfig := &messaging_pb.TopicConfiguration{
2020-04-30 10:05:34 +00:00
// IsTransient: true,
2020-04-18 05:39:21 +00:00
}
2020-04-19 07:18:32 +00:00
// send init response
initResponse := &messaging_pb.PublishResponse{
Config: nil,
Redirect: nil,
}
err = stream.Send(initResponse)
if err != nil {
return err
}
if initResponse.Redirect != nil {
return nil
}
2020-04-18 05:39:21 +00:00
2020-04-18 08:12:01 +00:00
// get lock
tp := TopicPartition{
Namespace: in.Init.Namespace,
Topic: in.Init.Topic,
Partition: in.Init.Partition,
}
2020-04-19 07:18:32 +00:00
tl := broker.topicLocks.RequestLock(tp, topicConfig, true)
2020-04-18 08:12:01 +00:00
defer broker.topicLocks.ReleaseLock(tp, true)
updatesChan := make(chan int32)
2020-04-18 05:39:21 +00:00
go func() {
for update := range updatesChan {
if err := stream.Send(&messaging_pb.PublishResponse{
Config: &messaging_pb.PublishResponse_ConfigMessage{
PartitionCount: update,
},
}); err != nil {
glog.V(0).Infof("err sending publish response: %v", err)
return
}
}
}()
2020-04-18 08:12:01 +00:00
// process each message
2020-04-18 05:39:21 +00:00
for {
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
2020-04-18 08:12:01 +00:00
if in.Data == nil {
continue
}
// fmt.Printf("received: %d : %s\n", len(in.Data.Value), string(in.Data.Value))
data, err := proto.Marshal(in.Data)
2020-04-18 05:39:21 +00:00
if err != nil {
glog.Errorf("marshall error: %v\n", err)
continue
}
2020-04-19 07:18:32 +00:00
tl.logBuffer.AddToBuffer(in.Data.Key, data)
2020-04-18 05:39:21 +00:00
}
}