Method Fetch

js.go:2968–3194

Fetch pulls a batch of messages from a stream for a pull consumer.

(batch int, opts ...PullOpt)

// Fetch pulls a batch of messages from a stream for a pull consumer.
func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
	if sub == nil {
		return nil, ErrBadSubscription
	}
	if batch < 1 {
		return nil, ErrInvalidArg
	}

	var o pullOpts
	for _, opt := range opts {
		if err := opt.configurePull(&o); err != nil {
			return nil, err
		}
	}
	if o.ctx != nil && o.ttl != 0 {
		return nil, ErrContextAndTimeout
	}

	sub.mu.Lock()
	jsi := sub.jsi
	// Reject if this is not a pull subscription. Note that sub.typ is SyncSubscription,
	// so check for jsi.pull boolean instead.
	if jsi == nil || !jsi.pull {
		sub.mu.Unlock()
		return nil, ErrTypeSubscription
	}

	nc := sub.conn
	nms := sub.jsi.nms
	rply, _ := newFetchInbox(jsi.deliver)
	js := sub.jsi.js
	pmc := len(sub.mch) > 0

	// All fetch requests have an expiration, in case of no explicit expiration
	// then the default timeout of the JetStream context is used.
	ttl := o.ttl
	if ttl == 0 {
		ttl = js.opts.wait
	}
	sub.mu.Unlock()

	// Use the given context or setup a default one for the span
	// of the pull batch request.
	var (
		ctx    = o.ctx
		err    error
		cancel context.CancelFunc
	)
	if ctx == nil {
		ctx, cancel = context.WithTimeout(context.Background(), ttl)
	} else if _, hasDeadline := ctx.Deadline(); !hasDeadline {
		// Prevent from passing the background context which will just block
		// and cannot be canceled either.
		if octx, ok := ctx.(ContextOpt); ok && octx.Context == context.Background() {
			return nil, ErrNoDeadlineContext
		}

		// If the context did not have a deadline, then create a new child context