Method fetch

jetstream/pull.go:899–1002

(req *pullRequest)

}

func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
	res := &fetchResult{
		msgs: make(chan Msg, req.Batch),
	}
	msgs := make(chan *nats.Msg, 2*req.Batch)
	subject := p.js.apiSubject(fmt.Sprintf(apiRequestNextT, p.stream, p.name))

	sub := &pullSubscription{
		consumer: p,
		done:     make(chan struct{}, 1),
		msgs:     msgs,
		errs:     make(chan error, 10),
	}
	inbox := p.js.conn.NewInbox()
	var err error
	sub.subscription, err = p.js.conn.ChanSubscribe(inbox, sub.msgs)
	if err != nil {
		return nil, err
	}
	req.PinID = p.getPinID()
	if err := sub.pull(req, subject); err != nil {
		return nil, err
	}

	var receivedMsgs, receivedBytes int
	hbTimer := sub.scheduleHeartbeatCheck(req.Heartbeat)

	// Use context if provided
	var ctxDone <-chan struct{}
	if req.ctx != nil {
		ctxDone = req.ctx.Done()
	}

	go func(res *fetchResult) {
		defer sub.subscription.Unsubscribe()
		defer close(res.msgs)
		for {
			select {
			case msg := <-msgs:
				res.Lock()
				if hbTimer != nil {
					hbTimer.Reset(2 * req.Heartbeat)
				}
				userMsg, err := checkMsg(msg)
				if err != nil {
					errNotTimeoutOrNoMsgs := !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, ErrNoMessages)
					if errNotTimeoutOrNoMsgs && !errors.Is(err, ErrMaxBytesExceeded) {
						res.err = err
					}
					if errors.Is(err, ErrPinIDMismatch) {
						p.setPinID("")
					}
					res.done = true
					res.Unlock()
					return
				}
				if !userMsg {
					res.Unlock()