Method waitForMsgs

nats.go:3571–3660

waitForMsgs waits on the conditional shared with readLoop and processMsg. It is used to deliver messages to asynchronous subscribers.

(s *Subscription)

// waitForMsgs waits on the conditional shared with readLoop and processMsg.
// It is used to deliver messages to asynchronous subscribers.
func (nc *Conn) waitForMsgs(s *Subscription) {
	var closed bool
	var delivered, max uint64

	// Used to account for adjustments to sub.pBytes when we wrap back around.
	msgLen := -1

	for {
		s.mu.Lock()
		// Do accounting for last msg delivered here so we only lock once
		// and drain state trips after callback has returned.
		if msgLen >= 0 {
			s.pMsgs--
			s.pBytes -= msgLen
			msgLen = -1
		}

		if s.pHead == nil && !s.closed {
			s.pCond.Wait()
		}
		// Pop the msg off the list
		m := s.pHead
		if m != nil {
			s.pHead = m.next
			if s.pHead == nil {
				s.pTail = nil
			}
			if m.barrier != nil {
				s.mu.Unlock()
				if atomic.AddInt64(&m.barrier.refs, -1) == 0 {
					m.barrier.f()
				}
				continue
			}
			msgLen = len(m.Data)
		}
		mcb := s.mcb
		max = s.max
		closed = s.closed
		var fcReply string
		if !s.closed {
			s.delivered++
			delivered = s.delivered
			if s.jsi != nil {
				fcReply = s.checkForFlowControlResponse()
			}
		}
		s.mu.Unlock()

		// Respond to flow control if applicable
		if fcReply != _EMPTY_ {
			nc.Publish(fcReply, nil)
		}

		if closed {
			break
		}