Method Messages

jetstream/pull.go:504–582

Messages returns MessagesContext, allowing continuously iterating over messages on a stream. Messages cannot be used concurrently when using ordered consumer. See [Consumer.Messages] for more details.

(opts ...PullMessagesOpt)

//
// See [Consumer.Messages] for more details.
func (p *pullConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, error) {
	consumeOpts, err := parseMessagesOpts(false, opts...)
	if err != nil {
		return nil, fmt.Errorf("%w: %s", ErrInvalidOption, err)
	}

	if len(p.info.Config.PriorityGroups) != 0 {
		if consumeOpts.Group == "" {
			return nil, fmt.Errorf("%w: %s", ErrInvalidOption, "priority group is required for priority consumer")
		}

		if !slices.Contains(p.info.Config.PriorityGroups, consumeOpts.Group) {
			return nil, fmt.Errorf("%w: %s", ErrInvalidOption, "invalid priority group")
		}
	} else if consumeOpts.Group != "" {
		return nil, fmt.Errorf("%w: %s", ErrInvalidOption, "priority group is not supported for this consumer")
	}

	p.Lock()
	subject := p.js.apiSubject(fmt.Sprintf(apiRequestNextT, p.stream, p.name))

	msgs := make(chan *nats.Msg, consumeOpts.MaxMessages)

	consumeID := nuid.Next()
	sub := &pullSubscription{
		id:          consumeID,
		consumer:    p,
		done:        make(chan struct{}, 1),
		msgs:        msgs,
		errs:        make(chan error, 10),
		fetchNext:   make(chan *pullRequest, 1),
		consumeOpts: consumeOpts,
	}
	sub.connStatusChanged = p.js.conn.StatusChanged(nats.CONNECTED, nats.RECONNECTING)
	inbox := p.js.conn.NewInbox()
	sub.subscription, err = p.js.conn.ChanSubscribe(inbox, sub.msgs)
	if err != nil {
		p.Unlock()
		return nil, err
	}
	sub.subscription.SetClosedHandler(func(sid string) func(string) {
		return func(subject string) {
			if sub.draining.Load() != 1 {
				// if we're not draining, subscription can be closed as soon
				// as closed handler is called
				// otherwise, we need to wait until all messages are drained
				// in Next
				p.subs.Delete(sid)
			}
			sub.closeMsgs()
		}
	}(sub.id))

	p.subs.Store(sub.id, sub)
	p.Unlock()

	go sub.pullMessages(subject)