use sync.Map instead of map and mutex for active message tracking
@@ -49,8 +49,7 @@ type HandlerFn func(ctx context.Context, msgs []*azservicebus.ReceivedMessage) (
// Subscription is an object that manages a subscription to an Azure Service Bus receiver, for a topic or queue.
type Subscription struct {
entity string
- mu sync.RWMutex
- activeMessages map[int64]*azservicebus.ReceivedMessage
+ activeMessages sync.Map
activeOperationsChan chan struct{}
requireSessions bool // read-only once set
timeout time.Duration
@@ -99,7 +98,6 @@ func NewSubscription(
s := &Subscription{
entity: opts.Entity,
- activeMessages: make(map[int64]*azservicebus.ReceivedMessage),
timeout: time.Duration(opts.TimeoutInSec) * time.Second,
maxBulkSubCount: *opts.MaxBulkSubCount,
requireSessions: opts.RequireSessions,
@@ -361,14 +359,11 @@ func (s *Subscription) RenewLocksBlocking(ctx context.Context, receiver Receiver
s.logger.Debugf("Renewed session %s locks for %s", sessionReceiver.SessionID(), s.entity)
} else {
// Snapshot the messages to try to renew locks for.
- s.mu.RLock()
- msgs := make([]*azservicebus.ReceivedMessage, 0, len(s.activeMessages))
- var i int
- for _, m := range s.activeMessages {
- msgs[i] = m
- i++
- }
- s.mu.RUnlock()
+ msgs := make([]*azservicebus.ReceivedMessage, 0)
+ s.activeMessages.Range(func(key, value interface{}) bool {
+ msgs = append(msgs, value.(*azservicebus.ReceivedMessage))
+ return true
+ })
if len(msgs) == 0 {
s.logger.Debugf("No active messages require lock renewal for %s", s.entity)
@@ -485,15 +480,11 @@ func (s *Subscription) addActiveMessage(m *azservicebus.ReceivedMessage) error {
logSuffix = fmt.Sprintf(" with session id %s", *m.SessionID)
}
s.logger.Debugf("Adding message %s with sequence number %d to active messages on %s%s", m.MessageID, *m.SequenceNumber, s.entity, logSuffix)
- s.mu.Lock()
- s.activeMessages[*m.SequenceNumber] = m
- s.mu.Unlock()
+ s.activeMessages.Store(*m.SequenceNumber, m)
return nil
}
func (s *Subscription) removeActiveMessage(messageID string, messageKey int64) {
s.logger.Debugf("Removing message %s with sequence number %d from active messages on %s", messageID, messageKey, s.entity)
- s.mu.Lock()
- delete(s.activeMessages, messageKey)
- s.mu.Unlock()
+ s.activeMessages.Delete(messageKey)
}
