gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into mq

Browse Source
This commit is contained in:
chrislu
2024-06-14 11:42:05 -07:00
parent 70b31e3a28 8aae82dd71
commit 32dbb6a2ce
26 changed files with 106 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
weed/mq/broker/broker_server.go
View File

@@ -2,6 +2,7 @@ package broker
import (
"github.com/seaweedfs/seaweedfs/weed/filer_client"
"context"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mq/pub_balancer"
"github.com/seaweedfs/seaweedfs/weed/mq/sub_coordinator"
@@ -75,9 +76,9 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
mqBroker.MasterClient.SetOnPeerUpdateFn(mqBroker.OnBrokerUpdate)
pubBalancer.OnPartitionChange = mqBroker.SubCoordinator.OnPartitionChange
go mqBroker.MasterClient.KeepConnectedToMaster()
go mqBroker.MasterClient.KeepConnectedToMaster(context.Background())
existingNodes := cluster.ListExistingPeerUpdates(mqBroker.MasterClient.GetMaster(), grpcDialOption, option.FilerGroup, cluster.FilerType)
existingNodes := cluster.ListExistingPeerUpdates(mqBroker.MasterClient.GetMaster(context.Background()), grpcDialOption, option.FilerGroup, cluster.FilerType)
for _, newNode := range existingNodes {
mqBroker.OnBrokerUpdate(newNode, time.Now())
}