seaweedfs/weed/messaging/broker/broker_server.go

115 lines
3.2 KiB
Go
Raw Normal View History

2020-04-18 16:12:01 +08:00
package broker
2020-03-04 16:39:47 +08:00
import (
"context"
"time"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/glog"
2020-03-04 16:39:47 +08:00
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
)
type MessageBrokerOption struct {
Filers []pb.ServerAddress
2020-03-04 16:39:47 +08:00
DefaultReplication string
MaxMB int
2020-05-05 17:05:28 +08:00
Ip string
2020-03-04 16:39:47 +08:00
Port int
2020-04-17 17:29:00 +08:00
Cipher bool
2020-03-04 16:39:47 +08:00
}
type MessageBroker struct {
option *MessageBrokerOption
grpcDialOption grpc.DialOption
2020-05-12 23:48:00 +08:00
topicManager *TopicManager
2020-03-04 16:39:47 +08:00
}
2020-04-17 17:29:00 +08:00
func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOption) (messageBroker *MessageBroker, err error) {
2020-03-04 16:39:47 +08:00
messageBroker = &MessageBroker{
option: option,
2020-04-17 17:29:00 +08:00
grpcDialOption: grpcDialOption,
2020-03-04 16:39:47 +08:00
}
2020-05-12 23:48:00 +08:00
messageBroker.topicManager = NewTopicManager(messageBroker)
2020-04-19 15:18:32 +08:00
2020-05-08 17:47:22 +08:00
messageBroker.checkFilers()
2020-05-05 17:05:28 +08:00
go messageBroker.keepConnectedToOneFiler()
2020-03-04 16:39:47 +08:00
return messageBroker, nil
}
2020-05-05 17:05:28 +08:00
func (broker *MessageBroker) keepConnectedToOneFiler() {
2020-03-04 16:39:47 +08:00
for {
for _, filer := range broker.option.Filers {
2020-05-05 17:05:28 +08:00
broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
stream, err := client.KeepConnected(ctx)
if err != nil {
2020-05-05 17:05:28 +08:00
glog.V(0).Infof("%s:%d failed to keep connected to %s: %v", broker.option.Ip, broker.option.Port, filer, err)
return err
}
2020-05-08 17:47:22 +08:00
initRequest := &filer_pb.KeepConnectedRequest{
Name: broker.option.Ip,
GrpcPort: uint32(broker.option.Port),
}
2020-05-12 23:48:00 +08:00
for _, tp := range broker.topicManager.ListTopicPartitions() {
2020-05-08 17:47:22 +08:00
initRequest.Resources = append(initRequest.Resources, tp.String())
}
if err := stream.Send(&filer_pb.KeepConnectedRequest{
Name: broker.option.Ip,
GrpcPort: uint32(broker.option.Port),
}); err != nil {
glog.V(0).Infof("broker %s:%d failed to init at %s: %v", broker.option.Ip, broker.option.Port, filer, err)
return err
}
// TODO send events of adding/removing topics
2020-05-05 17:05:28 +08:00
glog.V(0).Infof("conntected with filer: %v", filer)
for {
if err := stream.Send(&filer_pb.KeepConnectedRequest{
Name: broker.option.Ip,
GrpcPort: uint32(broker.option.Port),
}); err != nil {
glog.V(0).Infof("%s:%d failed to sendto %s: %v", broker.option.Ip, broker.option.Port, filer, err)
return err
}
// println("send heartbeat")
if _, err := stream.Recv(); err != nil {
glog.V(0).Infof("%s:%d failed to receive from %s: %v", broker.option.Ip, broker.option.Port, filer, err)
return err
}
// println("received reply")
2020-05-08 17:47:22 +08:00
time.Sleep(11 * time.Second)
2020-05-05 17:05:28 +08:00
// println("woke up")
}
return nil
2020-03-04 16:39:47 +08:00
})
2020-05-08 17:47:22 +08:00
time.Sleep(3 * time.Second)
2020-03-04 16:39:47 +08:00
}
}
}
func (broker *MessageBroker) withFilerClient(filer pb.ServerAddress, fn func(filer_pb.SeaweedFilerClient) error) error {
2020-03-04 16:39:47 +08:00
return pb.WithFilerClient(filer, broker.grpcDialOption, fn)
}
func (broker *MessageBroker) withMasterClient(master pb.ServerAddress, fn func(client master_pb.SeaweedClient) error) error {
2020-03-04 16:39:47 +08:00
return pb.WithMasterClient(master, broker.grpcDialOption, func(client master_pb.SeaweedClient) error {
return fn(client)
})
}