lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-08-24 05:35:11 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

go fmt

Browse Source
This commit is contained in:
chrislu 2024-05-23 08:25:53 -07:00
parent 77a7c5c2a0
commit cdeaaf95b4
5 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -44,9 +44,9 @@ type MessageQueueBroker struct {
currentFiler pb.ServerAddress currentFiler pb.ServerAddress
localTopicManager *topic.LocalTopicManager localTopicManager *topic.LocalTopicManager
PubBalancer *pub_balancer.PubBalancer PubBalancer *pub_balancer.PubBalancer
lockAsBalancer *cluster.LiveLock lockAsBalancer *cluster.LiveLock
SubCoordinator *sub_coordinator.SubCoordinator SubCoordinator *sub_coordinator.SubCoordinator
accessLock sync.Mutex accessLock sync.Mutex
fca *sub_coordinator.FilerClientAccessor fca *sub_coordinator.FilerClientAccessor
} }

8
weed/mq/client/cmd/weed_sub_kv/subscriber_kv.go
View File

@ -14,10 +14,10 @@ import (
) )
var ( var (
namespace = flag.String("ns", "test", "namespace") namespace = flag.String("ns", "test", "namespace")
t = flag.String("topic", "test", "topic") t = flag.String("topic", "test", "topic")
seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers") seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers")
maxPartitionCount = flag.Int("maxPartitionCount", 3, "max partition count") maxPartitionCount = flag.Int("maxPartitionCount", 3, "max partition count")
perPartitionConcurrency = flag.Int("perPartitionConcurrency", 1, "per partition concurrency") perPartitionConcurrency = flag.Int("perPartitionConcurrency", 1, "per partition concurrency")
clientId = flag.Uint("client_id", uint(util.RandomInt32()), "client id") clientId = flag.Uint("client_id", uint(util.RandomInt32()), "client id")

8
weed/mq/client/cmd/weed_sub_record/subscriber_record.go
View File

@ -16,10 +16,10 @@ import (
) )
var ( var (
namespace = flag.String("ns", "test", "namespace") namespace = flag.String("ns", "test", "namespace")
t = flag.String("topic", "test", "topic") t = flag.String("topic", "test", "topic")
seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers") seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers")
maxPartitionCount = flag.Int("maxPartitionCount", 3, "max partition count") maxPartitionCount = flag.Int("maxPartitionCount", 3, "max partition count")
perPartitionConcurrency = flag.Int("perPartitionConcurrency", 1, "per partition concurrency") perPartitionConcurrency = flag.Int("perPartitionConcurrency", 1, "per partition concurrency")
clientId = flag.Uint("client_id", uint(util.RandomInt32()), "client id") clientId = flag.Uint("client_id", uint(util.RandomInt32()), "client id")

2
weed/mq/client/sub_client/on_each_partition.go
View File

@ -65,7 +65,7 @@ func (sub *TopicSubscriber) onEachPartition(assigned *mq_pb.BrokerPartitionAssig
select { select {
case <-stopCh: case <-stopCh:
break break
case ack := <- partitionOffsetChan: case ack := <-partitionOffsetChan:
subscribeClient.SendMsg(&mq_pb.SubscribeMessageRequest{ subscribeClient.SendMsg(&mq_pb.SubscribeMessageRequest{
Message: &mq_pb.SubscribeMessageRequest_Ack{ Message: &mq_pb.SubscribeMessageRequest_Ack{
Ack: &mq_pb.SubscribeMessageRequest_AckMessage{ Ack: &mq_pb.SubscribeMessageRequest_AckMessage{

14
weed/mq/client/sub_client/subscriber.go
View File

@ -27,16 +27,16 @@ type OnEachMessageFunc func(key, value []byte) (err error)
type OnCompletionFunc func() type OnCompletionFunc func()
type TopicSubscriber struct { type TopicSubscriber struct {
SubscriberConfig *SubscriberConfiguration SubscriberConfig *SubscriberConfiguration
ContentConfig *ContentConfiguration ContentConfig *ContentConfiguration
brokerPartitionAssignmentChan chan *mq_pb.SubscriberToSubCoordinatorResponse brokerPartitionAssignmentChan chan *mq_pb.SubscriberToSubCoordinatorResponse
brokerPartitionAssignmentAckChan chan *mq_pb.SubscriberToSubCoordinatorRequest brokerPartitionAssignmentAckChan chan *mq_pb.SubscriberToSubCoordinatorRequest
OnEachMessageFunc OnEachMessageFunc OnEachMessageFunc OnEachMessageFunc
OnCompletionFunc OnCompletionFunc OnCompletionFunc OnCompletionFunc
bootstrapBrokers []string bootstrapBrokers []string
waitForMoreMessage bool waitForMoreMessage bool
activeProcessors map[topic.Partition]*ProcessorState activeProcessors map[topic.Partition]*ProcessorState
activeProcessorsLock sync.Mutex activeProcessorsLock sync.Mutex
} }
func NewTopicSubscriber(bootstrapBrokers []string, subscriber *SubscriberConfiguration, content *ContentConfiguration) *TopicSubscriber { func NewTopicSubscriber(bootstrapBrokers []string, subscriber *SubscriberConfiguration, content *ContentConfiguration) *TopicSubscriber {