Method drainConnection

nats.go:6077–6173

drainConnection will run in a separate Go routine and will flush all publishes and drain all active subscriptions.

()

// drainConnection will run in a separate Go routine and will
// flush all publishes and drain all active subscriptions.
func (nc *Conn) drainConnection() {
	// Snapshot subs list.
	nc.mu.Lock()

	// Check again here if we are in a state to not process.
	if nc.isClosed() {
		nc.mu.Unlock()
		return
	}
	if nc.isConnecting() || nc.isReconnecting() {
		nc.mu.Unlock()
		// Move to closed state.
		nc.Close()
		return
	}

	subs := make([]*Subscription, 0, len(nc.subs))
	for _, s := range nc.subs {
		if s == nc.respMux {
			// Skip since might be in use while messages
			// are being processed (can miss responses).
			continue
		}
		subs = append(subs, s)
	}
	errCB := nc.Opts.AsyncErrorCB
	drainWait := nc.Opts.DrainTimeout
	respMux := nc.respMux
	nc.mu.Unlock()

	// for pushing errors with context.
	pushErr := func(err error) {
		nc.mu.Lock()
		nc.err = err
		if errCB != nil {
			nc.ach.push(func() { errCB(nc, nil, err) })
		}
		nc.mu.Unlock()
	}

	// Do subs first, skip request handler if present.
	for _, s := range subs {
		if err := s.Drain(); err != nil {
			// We will notify about these but continue.
			pushErr(err)
		}
	}

	// Wait for the subscriptions to drop to zero.
	timeout := time.Now().Add(drainWait)
	var min int
	if respMux != nil {
		min = 1
	} else {
		min = 0
	}
	for time.Now().Before(timeout) {
		if nc.NumSubscriptions() == min {