seaweedfs/weed/messaging/msg_broker_grpc_server.go

104 lines
2.4 KiB
Go
Raw Normal View History

2020-04-16 09:55:09 +00:00
package messaging
import (
"context"
"fmt"
"io"
"time"
"github.com/golang/protobuf/proto"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
"github.com/chrislusf/seaweedfs/weed/util/log_buffer"
)
2020-04-16 10:29:57 +00:00
func (broker *MessageBroker) Subscribe(server messaging_pb.SeaweedMessaging_SubscribeServer) error {
2020-04-16 09:55:09 +00:00
panic("implement me")
}
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
}
2020-04-16 10:29:57 +00:00
namespace, topic, partition := in.Init.Namespace, in.Init.Topic, in.Init.Partition
2020-04-16 09:55:09 +00:00
updatesChan := make(chan int32)
2020-04-17 09:29:38 +00:00
// TODO look it up
topicConfig := &messaging_pb.TopicConfiguration{
}
2020-04-16 09:55:09 +00:00
go func() {
for update := range updatesChan {
if err := stream.Send(&messaging_pb.PublishResponse{
2020-04-16 10:29:57 +00:00
Config: &messaging_pb.PublishResponse_ConfigMessage{
PartitionCount: update,
},
2020-04-16 09:55:09 +00:00
}); err != nil {
glog.V(0).Infof("err sending publish response: %v", err)
return
}
}
}()
logBuffer := log_buffer.NewLogBuffer(time.Minute, func(startTime, stopTime time.Time, buf []byte) {
2020-04-17 09:29:38 +00:00
targetFile := fmt.Sprintf(
"%s/%s/%s/%04d-%02d-%02d/%02d-%02d.part%02d",
2020-04-16 09:55:09 +00:00
filer2.TopicsDir, namespace, topic,
startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(),
partition,
)
2020-04-17 09:29:38 +00:00
if err := broker.appendToFile(targetFile, topicConfig, buf); err != nil {
glog.V(0).Infof("log write failed %s: %v", targetFile, err)
}
2020-04-16 09:55:09 +00:00
}, func() {
// notify subscribers
})
for {
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
m := &messaging_pb.Message{
Timestamp: time.Now().UnixNano(),
2020-04-16 10:29:57 +00:00
Key: in.Data.Key,
Value: in.Data.Value,
Headers: in.Data.Headers,
2020-04-16 09:55:09 +00:00
}
data, err := proto.Marshal(m)
if err != nil {
glog.Errorf("marshall error: %v\n", err)
continue
}
2020-04-16 10:29:57 +00:00
logBuffer.AddToBuffer(in.Data.Key, data)
2020-04-16 09:55:09 +00:00
}
}
func (broker *MessageBroker) ConfigureTopic(c context.Context, request *messaging_pb.ConfigureTopicRequest) (*messaging_pb.ConfigureTopicResponse, error) {
panic("implement me")
}
func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *messaging_pb.GetTopicConfigurationRequest) (*messaging_pb.GetTopicConfigurationResponse, error) {
panic("implement me")
}