2020-05-05 09:05:28 +00:00
|
|
|
package msgclient
|
2020-04-18 19:44:55 +00:00
|
|
|
|
2020-04-18 22:17:27 +00:00
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
|
2020-04-21 07:59:55 +00:00
|
|
|
"github.com/OneOfOne/xxhash"
|
2020-05-08 09:47:22 +00:00
|
|
|
"google.golang.org/grpc"
|
2020-04-21 07:59:55 +00:00
|
|
|
|
2020-05-05 09:05:28 +00:00
|
|
|
"github.com/chrislusf/seaweedfs/weed/messaging/broker"
|
2020-04-18 22:17:27 +00:00
|
|
|
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
|
|
|
|
)
|
2020-04-18 19:44:55 +00:00
|
|
|
|
|
|
|
type Publisher struct {
|
2020-04-21 07:59:55 +00:00
|
|
|
publishClients []messaging_pb.SeaweedMessaging_PublishClient
|
|
|
|
topicConfiguration *messaging_pb.TopicConfiguration
|
|
|
|
messageCount uint64
|
|
|
|
publisherId string
|
2020-04-18 22:17:27 +00:00
|
|
|
}
|
2020-05-10 10:50:30 +00:00
|
|
|
|
2020-04-21 07:59:55 +00:00
|
|
|
func (mc *MessagingClient) NewPublisher(publisherId, namespace, topic string) (*Publisher, error) {
|
|
|
|
// read topic configuration
|
|
|
|
topicConfiguration := &messaging_pb.TopicConfiguration{
|
|
|
|
PartitionCount: 4,
|
|
|
|
}
|
|
|
|
publishClients := make([]messaging_pb.SeaweedMessaging_PublishClient, topicConfiguration.PartitionCount)
|
|
|
|
for i := 0; i < int(topicConfiguration.PartitionCount); i++ {
|
2020-05-11 06:45:48 +00:00
|
|
|
tp := broker.TopicPartition{
|
2020-05-08 09:47:22 +00:00
|
|
|
Namespace: namespace,
|
|
|
|
Topic: topic,
|
|
|
|
Partition: int32(i),
|
2020-05-11 06:45:48 +00:00
|
|
|
}
|
|
|
|
grpcClientConn, err := mc.findBroker(tp)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
client, err := setupPublisherClient(grpcClientConn, tp)
|
2020-04-21 07:59:55 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
publishClients[i] = client
|
|
|
|
}
|
|
|
|
return &Publisher{
|
|
|
|
publishClients: publishClients,
|
|
|
|
topicConfiguration: topicConfiguration,
|
|
|
|
}, nil
|
|
|
|
}
|
|
|
|
|
2020-05-05 09:05:28 +00:00
|
|
|
func setupPublisherClient(grpcConnection *grpc.ClientConn, tp broker.TopicPartition) (messaging_pb.SeaweedMessaging_PublishClient, error) {
|
2020-04-18 22:17:27 +00:00
|
|
|
|
2020-05-05 09:05:28 +00:00
|
|
|
stream, err := messaging_pb.NewSeaweedMessagingClient(grpcConnection).Publish(context.Background())
|
2020-04-18 22:17:27 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// send init message
|
|
|
|
err = stream.Send(&messaging_pb.PublishRequest{
|
|
|
|
Init: &messaging_pb.PublishRequest_InitMessage{
|
2020-05-05 09:05:28 +00:00
|
|
|
Namespace: tp.Namespace,
|
|
|
|
Topic: tp.Topic,
|
|
|
|
Partition: tp.Partition,
|
2020-04-18 22:17:27 +00:00
|
|
|
},
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// process init response
|
|
|
|
initResponse, err := stream.Recv()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
if initResponse.Redirect != nil {
|
|
|
|
// TODO follow redirection
|
|
|
|
}
|
|
|
|
if initResponse.Config != nil {
|
|
|
|
}
|
|
|
|
|
|
|
|
// setup looks for control messages
|
|
|
|
doneChan := make(chan error, 1)
|
|
|
|
go func() {
|
|
|
|
for {
|
|
|
|
in, err := stream.Recv()
|
|
|
|
if err != nil {
|
|
|
|
doneChan <- err
|
|
|
|
return
|
|
|
|
}
|
2020-04-21 00:48:06 +00:00
|
|
|
if in.Redirect != nil {
|
2020-04-18 22:17:27 +00:00
|
|
|
}
|
2020-04-21 00:48:06 +00:00
|
|
|
if in.Config != nil {
|
2020-04-18 22:17:27 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2020-04-21 07:59:55 +00:00
|
|
|
return stream, nil
|
|
|
|
|
2020-04-18 19:44:55 +00:00
|
|
|
}
|
|
|
|
|
2020-04-20 06:37:04 +00:00
|
|
|
func (p *Publisher) Publish(m *messaging_pb.Message) error {
|
2020-04-21 07:59:55 +00:00
|
|
|
hashValue := p.messageCount
|
|
|
|
p.messageCount++
|
|
|
|
if p.topicConfiguration.Partitoning == messaging_pb.TopicConfiguration_NonNullKeyHash {
|
|
|
|
if m.Key != nil {
|
|
|
|
hashValue = xxhash.Checksum64(m.Key)
|
|
|
|
}
|
|
|
|
} else if p.topicConfiguration.Partitoning == messaging_pb.TopicConfiguration_KeyHash {
|
|
|
|
hashValue = xxhash.Checksum64(m.Key)
|
|
|
|
} else {
|
|
|
|
// round robin
|
|
|
|
}
|
2020-04-18 22:17:27 +00:00
|
|
|
|
2020-04-21 07:59:55 +00:00
|
|
|
idx := int(hashValue) % len(p.publishClients)
|
|
|
|
if idx < 0 {
|
|
|
|
idx += len(p.publishClients)
|
|
|
|
}
|
|
|
|
return p.publishClients[idx].Send(&messaging_pb.PublishRequest{
|
2020-04-18 22:17:27 +00:00
|
|
|
Data: m,
|
|
|
|
})
|
2020-04-18 19:44:55 +00:00
|
|
|
}
|