Method NextMsg

nats.go:5357–5425

NextMsg will return the next message available to a synchronous subscriber or block until one is available. An error is returned if the subscription is invalid (ErrBadSubscription), the connection is closed (ErrConnectionClosed), the timeout is reached (ErrTimeout), or if there were no responders (E

(timeout time.Duration)

// the connection is closed (ErrConnectionClosed), the timeout is reached (ErrTimeout),
// or if there were no responders (ErrNoResponders) when used in the context of a request/reply.
func (s *Subscription) NextMsg(timeout time.Duration) (*Msg, error) {
	if s == nil {
		return nil, ErrBadSubscription
	}

	s.mu.Lock()
	err := s.validateNextMsgState(false)
	if err != nil {
		s.mu.Unlock()
		return nil, err
	}

	// snapshot
	mch := s.mch
	s.mu.Unlock()

	var ok bool
	var msg *Msg

	// If something is available right away, let's optimize that case.
	select {
	case msg, ok = <-mch:
		if !ok {
			return nil, s.getNextMsgErr()
		}
		if err := s.processNextMsgDelivered(msg); err != nil {
			return nil, err
		} else {
			return msg, nil
		}
	default:
	}

	// If we are here a message was not immediately available, so lets loop
	// with a timeout.

	t := globalTimerPool.Get(timeout)
	defer globalTimerPool.Put(t)

	if s.errCh != nil {
		select {
		case msg, ok = <-mch:
			if !ok {
				return nil, s.getNextMsgErr()
			}
			if err := s.processNextMsgDelivered(msg); err != nil {
				return nil, err
			}
		case err := <-s.errCh:
			return nil, err
		case <-t.C:
			return nil, ErrTimeout
		}
	} else {
		select {
		case msg, ok = <-mch:
			if !ok {
				return nil, s.getNextMsgErr()