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

refactor

Browse Source
This commit is contained in:
chrislu
2024-08-10 12:39:16 -07:00
parent 7438648d1c
commit b6fd1ff4ce
2 changed files with 7 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
weed/mq/topic/local_manager.go
View File

@@ -28,10 +28,7 @@ func (manager *LocalTopicManager) AddLocalPartition(topic Topic, localPartition
if !manager.topics.SetIfAbsent(topic.String(), localTopic) { if !manager.topics.SetIfAbsent(topic.String(), localTopic) {
localTopic, _ = manager.topics.Get(topic.String()) localTopic, _ = manager.topics.Get(topic.String())
} }
if localTopic.findPartition(localPartition.Partition) != nil { localTopic.AddPartition(localPartition)
return
}
localTopic.Partitions = append(localTopic.Partitions, localPartition)
} }
// GetLocalPartition gets a topic from the local topic manager // GetLocalPartition gets a topic from the local topic manager

6
weed/mq/topic/local_topic.go
View File

@@ -37,6 +37,12 @@ func (localTopic *LocalTopic) removePartition(partition Partition) bool {
localTopic.Partitions = append(localTopic.Partitions[:foundPartitionIndex], localTopic.Partitions[foundPartitionIndex+1:]...) localTopic.Partitions = append(localTopic.Partitions[:foundPartitionIndex], localTopic.Partitions[foundPartitionIndex+1:]...)
return true return true
} }
func (localTopic *LocalTopic) AddPartition(localPartition *LocalPartition) {
if localTopic.findPartition(localPartition.Partition) != nil {
return
}
localTopic.Partitions = append(localTopic.Partitions, localPartition)
}
func (localTopic *LocalTopic) closePartitionPublishers(unixTsNs int64) bool { func (localTopic *LocalTopic) closePartitionPublishers(unixTsNs int64) bool {
var wg sync.WaitGroup var wg sync.WaitGroup