Method Consume

jetstream/pull.go:202–426

Consume can be used to continuously receive messages and handle them with the provided callback function. Consume cannot be used concurrently when using ordered consumer. See [Consumer.Consume] for more details.

(handler MessageHandler, opts ...PullConsumeOpt)

//
// See [Consumer.Consume] for more details.
func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error) {
	if handler == nil {
		return nil, ErrHandlerRequired
	}
	consumeOpts, err := parseConsumeOpts(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))

	consumeID := nuid.Next()
	sub := &pullSubscription{
		id:          consumeID,
		consumer:    p,
		errs:        make(chan error, 10),
		done:        make(chan struct{}, 1),
		fetchNext:   make(chan *pullRequest, 1),
		consumeOpts: consumeOpts,
	}
	sub.connStatusChanged = p.js.conn.StatusChanged(nats.CONNECTED, nats.RECONNECTING, nats.CLOSED)

	sub.hbMonitor = sub.scheduleHeartbeatCheck(consumeOpts.Heartbeat)

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

	internalHandler := func(msg *nats.Msg) {
		if sub.hbMonitor != nil {
			sub.hbMonitor.Stop()
		}
		userMsg, msgErr := checkMsg(msg)
		if !userMsg && msgErr == nil {
			if sub.hbMonitor != nil {
				sub.hbMonitor.Reset(2 * consumeOpts.Heartbeat)
			}
			return
		}
		if !userMsg {
			// heartbeat message
			if msgErr == nil {
				return
			}