1 files changed, 28 insertions, 23 deletions

diff --git a/weed/mq/broker/broker_server.go b/weed/mq/broker/broker_server.go
index 94888e14c..52b34ddbc 100644
--- a/weed/mq/broker/broker_server.go
+++ b/weed/mq/broker/broker_server.go
@@ -3,7 +3,8 @@ package broker
 import (
 	"fmt"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
-	"github.com/seaweedfs/seaweedfs/weed/mq/balancer"
+	"github.com/seaweedfs/seaweedfs/weed/mq/pub_balancer"
+	"github.com/seaweedfs/seaweedfs/weed/mq/sub_coordinator"
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
 	"time"
 
@@ -37,20 +38,24 @@ type MessageQueueBroker struct {
 	filers            map[pb.ServerAddress]struct{}
 	currentFiler      pb.ServerAddress
 	localTopicManager *topic.LocalTopicManager
-	Balancer          *balancer.Balancer
+	Balancer          *pub_balancer.Balancer
 	lockAsBalancer    *cluster.LiveLock
 	currentBalancer   pb.ServerAddress
+	Coordinator       *sub_coordinator.Coordinator
 }
 
 func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (mqBroker *MessageQueueBroker, err error) {
 
+	pub_broker_balancer := pub_balancer.NewBalancer()
+
 	mqBroker = &MessageQueueBroker{
 		option:            option,
 		grpcDialOption:    grpcDialOption,
 		MasterClient:      wdclient.NewMasterClient(grpcDialOption, option.FilerGroup, cluster.BrokerType, pb.NewServerAddress(option.Ip, option.Port, 0), option.DataCenter, option.Rack, *pb.NewServiceDiscoveryFromMap(option.Masters)),
 		filers:            make(map[pb.ServerAddress]struct{}),
 		localTopicManager: topic.NewLocalTopicManager(),
-		Balancer:          balancer.NewBalancer(),
+		Balancer:          pub_broker_balancer,
+		Coordinator:       sub_coordinator.NewCoordinator(pub_broker_balancer),
 	}
 	mqBroker.MasterClient.SetOnPeerUpdateFn(mqBroker.OnBrokerUpdate)
 
@@ -67,10 +72,10 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
 			time.Sleep(time.Millisecond * 237)
 		}
 		self := fmt.Sprintf("%s:%d", option.Ip, option.Port)
-		glog.V(1).Infof("broker %s found filer %s", self, mqBroker.currentFiler)
+		glog.V(0).Infof("broker %s found filer %s", self, mqBroker.currentFiler)
 
 		lockClient := cluster.NewLockClient(grpcDialOption, mqBroker.currentFiler)
-		mqBroker.lockAsBalancer = lockClient.StartLock(balancer.LockBrokerBalancer, self)
+		mqBroker.lockAsBalancer = lockClient.StartLock(pub_balancer.LockBrokerBalancer, self)
 		for {
 			err := mqBroker.BrokerConnectToBalancer(self)
 			if err != nil {
@@ -83,22 +88,22 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
 	return mqBroker, nil
 }
 
-func (broker *MessageQueueBroker) OnBrokerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
+func (b *MessageQueueBroker) OnBrokerUpdate(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
 	if update.NodeType != cluster.FilerType {
 		return
 	}
 
 	address := pb.ServerAddress(update.Address)
 	if update.IsAdd {
-		broker.filers[address] = struct{}{}
-		if broker.currentFiler == "" {
-			broker.currentFiler = address
+		b.filers[address] = struct{}{}
+		if b.currentFiler == "" {
+			b.currentFiler = address
 		}
 	} else {
-		delete(broker.filers, address)
-		if broker.currentFiler == address {
-			for filer := range broker.filers {
-				broker.currentFiler = filer
+		delete(b.filers, address)
+		if b.currentFiler == address {
+			for filer := range b.filers {
+				b.currentFiler = filer
 				break
 			}
 		}
@@ -106,39 +111,39 @@ func (broker *MessageQueueBroker) OnBrokerUpdate(update *master_pb.ClusterNodeUp
 
 }
 
-func (broker *MessageQueueBroker) GetFiler() pb.ServerAddress {
-	return broker.currentFiler
+func (b *MessageQueueBroker) GetFiler() pb.ServerAddress {
+	return b.currentFiler
 }
 
-func (broker *MessageQueueBroker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+func (b *MessageQueueBroker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	return pb.WithFilerClient(streamingMode, 0, broker.GetFiler(), broker.grpcDialOption, fn)
+	return pb.WithFilerClient(streamingMode, 0, b.GetFiler(), b.grpcDialOption, fn)
 
 }
 
-func (broker *MessageQueueBroker) AdjustedUrl(location *filer_pb.Location) string {
+func (b *MessageQueueBroker) AdjustedUrl(location *filer_pb.Location) string {
 
 	return location.Url
 
 }
 
-func (broker *MessageQueueBroker) GetDataCenter() string {
+func (b *MessageQueueBroker) GetDataCenter() string {
 
 	return ""
 
 }
 
-func (broker *MessageQueueBroker) withMasterClient(streamingMode bool, master pb.ServerAddress, fn func(client master_pb.SeaweedClient) error) error {
+func (b *MessageQueueBroker) withMasterClient(streamingMode bool, master pb.ServerAddress, fn func(client master_pb.SeaweedClient) error) error {
 
-	return pb.WithMasterClient(streamingMode, master, broker.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+	return pb.WithMasterClient(streamingMode, master, b.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
 		return fn(client)
 	})
 
 }
 
-func (broker *MessageQueueBroker) withBrokerClient(streamingMode bool, server pb.ServerAddress, fn func(client mq_pb.SeaweedMessagingClient) error) error {
+func (b *MessageQueueBroker) withBrokerClient(streamingMode bool, server pb.ServerAddress, fn func(client mq_pb.SeaweedMessagingClient) error) error {
 
-	return pb.WithBrokerGrpcClient(streamingMode, server.String(), broker.grpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
+	return pb.WithBrokerGrpcClient(streamingMode, server.String(), b.grpcDialOption, func(client mq_pb.SeaweedMessagingClient) error {
 		return fn(client)
 	})