Method processNextMsgDelivered

nats.go:5539–5578

processNextMsgDelivered takes a message and applies the needed accounting to the stats from the subscription, returning an error in case we have the maximum number of messages have been delivered already. It should not be called while holding the lock.

(msg *Msg)

// error in case we have the maximum number of messages have been
// delivered already. It should not be called while holding the lock.
func (s *Subscription) processNextMsgDelivered(msg *Msg) error {
	s.mu.Lock()
	nc := s.conn
	max := s.max

	var fcReply string
	// Update some stats.
	s.delivered++
	delivered := s.delivered
	if s.jsi != nil {
		fcReply = s.checkForFlowControlResponse()
	}

	if s.typ == SyncSubscription {
		s.pMsgs--
		s.pBytes -= len(msg.Data)
	}
	s.mu.Unlock()

	if fcReply != _EMPTY_ {
		nc.Publish(fcReply, nil)
	}

	if max > 0 {
		if delivered > max {
			return ErrMaxMessages
		}
		// Remove subscription if we have reached max.
		if delivered == max {
			nc.mu.Lock()
			nc.removeSub(s)
			nc.mu.Unlock()
		}
	}
	if len(msg.Data) == 0 && msg.Header.Get(statusHdr) == noResponders {
		return ErrNoResponders
	}

	return nil
}

// Queued returns the number of queued messages in the client for this subscription.
//