Method resetOrderedConsumer

js.go:2199–2312

We are here if we have detected a gap with an ordered consumer. We will create a new consumer and rewire the low level subscription. Lock should be held.

(sseq uint64)

// We will create a new consumer and rewire the low level subscription.
// Lock should be held.
func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
	nc := sub.conn
	if sub.jsi == nil || nc == nil || sub.closed {
		return
	}

	var maxStr string
	// If there was an AUTO_UNSUB done, we need to adjust the new value
	// to send after the SUB for the new sid.
	if sub.max > 0 {
		if sub.jsi.fciseq < sub.max {
			adjustedMax := sub.max - sub.jsi.fciseq
			maxStr = strconv.Itoa(int(adjustedMax))
		} else {
			// We are already at the max, so we should just unsub the
			// existing sub and be done
			go func(sid int64) {
				nc.mu.Lock()
				nc.bw.appendString(fmt.Sprintf(unsubProto, sid, _EMPTY_))
				nc.kickFlusher()
				nc.mu.Unlock()
			}(sub.sid)
			return
		}
	}

	// Quick unsubscribe. Since we know this is a simple push subscriber we do in place.
	osid := sub.applyNewSID()

	// Grab new inbox.
	newDeliver := nc.NewInbox()
	sub.Subject = newDeliver

	// Snapshot the new sid under sub lock.
	nsid := sub.sid

	// We are still in the low level readLoop for the connection so we need
	// to spin a go routine to try to create the new consumer.
	go func() {
		// Unsubscribe and subscribe with new inbox and sid.
		// Remap a new low level sub into this sub since its client accessible.
		// This is done here in this go routine to prevent lock inversion.
		nc.mu.Lock()
		nc.bw.appendString(fmt.Sprintf(unsubProto, osid, _EMPTY_))
		nc.bw.appendString(fmt.Sprintf(subProto, newDeliver, _EMPTY_, nsid))
		if maxStr != _EMPTY_ {
			nc.bw.appendString(fmt.Sprintf(unsubProto, nsid, maxStr))
		}
		nc.kickFlusher()
		nc.mu.Unlock()

		pushErr := func(err error) {
			nc.handleConsumerSequenceMismatch(sub, fmt.Errorf("%w: recreating ordered consumer", err))
			nc.unsubscribe(sub, 0, true)
		}

		sub.mu.Lock()
		jsi := sub.jsi