75 lines
2 KiB
Diff
75 lines
2 KiB
Diff
# UNDF: UNDF-2026-000001086
|
|
--- a/pubsub/inmem.go
|
|
+++ b/pubsub/inmem.go
|
|
@@ -14,7 +14,6 @@ import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rs/zerolog/log"
|
|
- "golang.org/x/exp/slices"
|
|
)
|
|
|
|
@@ -91,7 +90,8 @@ func (r *InMemory) Publish(ctx context.Context, topic string, payload []byte, op
|
|
topic = formatTopic(pubConfig.app, pubConfig.namespace, topic)
|
|
wg := sync.WaitGroup{}
|
|
for _, sub := range r.registry {
|
|
- if slices.Contains(sub.topics, topic) && !sub.isClosed() {
|
|
+ if sub.hasTopic(topic) && !sub.isClosed() {
|
|
wg.Add(1)
|
|
go func(subscriber *inMemorySubscriber) {
|
|
defer wg.Done()
|
|
@@ -128,9 +128,9 @@ func (s *inMemorySubscriber) Subscribe(_ context.Context, topics ...string) erro
|
|
defer s.mutex.RUnlock()
|
|
topics = s.formatTopics(topics...)
|
|
for _, ch := range topics {
|
|
- if slices.Contains(s.topics, ch) {
|
|
- continue
|
|
- }
|
|
- s.topics = append(s.topics, ch)
|
|
+ if _, ok := s.topicSet[ch]; !ok {
|
|
+ s.topicSet[ch] = struct{}{}
|
|
+ s.topics = append(s.topics, ch)
|
|
+ }
|
|
}
|
|
return nil
|
|
}
|
|
@@ -140,9 +140,9 @@ func (s *inMemorySubscriber) Unsubscribe(_ context.Context, topics ...string) er
|
|
defer s.mutex.RUnlock()
|
|
topics = s.formatTopics(topics...)
|
|
for _, ch := range topics {
|
|
- if slices.Contains(s.topics, ch) {
|
|
- s.topics[i] = s.topics[len(s.topics)-1]
|
|
- s.topics = s.topics[:len(s.topics)-1]
|
|
- }
|
|
+ if _, ok := s.topicSet[ch]; ok {
|
|
+ delete(s.topicSet, ch)
|
|
+ // rebuild slice from set
|
|
+ s.topics = s.topics[:0]
|
|
+ for t := range s.topicSet {
|
|
+ s.topics = append(s.topics, t)
|
|
+ }
|
|
+ }
|
|
}
|
|
return nil
|
|
}
|
|
|
|
+func (s *inMemorySubscriber) hasTopic(topic string) bool {
|
|
+ s.mutex.RLock()
|
|
+ defer s.mutex.RUnlock()
|
|
+ _, ok := s.topicSet[topic]
|
|
+ return ok
|
|
+}
|
|
|
|
// struct change: add topicSet map alongside topics slice
|
|
type inMemorySubscriber struct {
|
|
config *SubscribeConfig
|
|
handler func([]byte) error
|
|
channel chan []byte
|
|
once sync.Once
|
|
mutex sync.RWMutex
|
|
topics []string
|
|
+ topicSet map[string]struct{}
|
|
closed bool
|
|
}
|