diff options
Diffstat (limited to 'weed/mq/broker/broker_server.go')
| -rw-r--r-- | weed/mq/broker/broker_server.go | 5 |
1 files changed, 4 insertions, 1 deletions
diff --git a/weed/mq/broker/broker_server.go b/weed/mq/broker/broker_server.go index 52b34ddbc..615964621 100644 --- a/weed/mq/broker/broker_server.go +++ b/weed/mq/broker/broker_server.go @@ -28,6 +28,7 @@ type MessageQueueBrokerOption struct { Ip string Port int Cipher bool + VolumeServerAccess string // how to access volume servers } type MessageQueueBroker struct { @@ -47,6 +48,7 @@ type MessageQueueBroker struct { func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (mqBroker *MessageQueueBroker, err error) { pub_broker_balancer := pub_balancer.NewBalancer() + coordinator := sub_coordinator.NewCoordinator(pub_broker_balancer) mqBroker = &MessageQueueBroker{ option: option, @@ -55,9 +57,10 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial filers: make(map[pb.ServerAddress]struct{}), localTopicManager: topic.NewLocalTopicManager(), Balancer: pub_broker_balancer, - Coordinator: sub_coordinator.NewCoordinator(pub_broker_balancer), + Coordinator: coordinator, } mqBroker.MasterClient.SetOnPeerUpdateFn(mqBroker.OnBrokerUpdate) + pub_broker_balancer.OnPartitionChange = mqBroker.Coordinator.OnPartitionChange go mqBroker.MasterClient.KeepConnectedToMaster() |