Method unsubscribe

nats.go:5291–5351

unsubscribe performs the low level unsubscribe to the server. Use Subscription.Unsubscribe()

(sub *Subscription, max int, drainMode bool)

// unsubscribe performs the low level unsubscribe to the server.
// Use Subscription.Unsubscribe()
func (nc *Conn) unsubscribe(sub *Subscription, max int, drainMode bool) error {
	var maxStr string
	if max > 0 {
		sub.mu.Lock()
		sub.max = uint64(max)
		if sub.delivered < sub.max {
			maxStr = strconv.Itoa(max)
		}
		sub.mu.Unlock()
	}

	nc.mu.Lock()
	// ok here, but defer is expensive
	defer nc.mu.Unlock()

	if nc.isClosed() {
		return ErrConnectionClosed
	}

	nc.subsMu.RLock()
	s := nc.subs[sub.sid]
	nc.subsMu.RUnlock()
	// Already unsubscribed
	if s == nil {
		return nil
	}

	if maxStr == _EMPTY_ && !drainMode {
		nc.removeSub(s)
	}

	if drainMode {
		s.mu.Lock()
		s.draining = true
		sub.changeSubStatus(SubscriptionDraining)
		s.mu.Unlock()
		go nc.checkDrained(sub)
	}

	// We will send these for all subs when we reconnect
	// so that we can suppress here.
	if !nc.isReconnecting() {
		nc.bw.appendString(fmt.Sprintf(unsubProto, s.sid, maxStr))
		nc.kickFlusher()
	}

	// For JetStream subscriptions cancel the attached context if there is any.
	var cancel func()
	sub.mu.Lock()
	jsi := sub.jsi
	if jsi != nil {
		cancel = jsi.cancel
		jsi.cancel = nil
	}
	sub.mu.Unlock()
	if cancel != nil {
		cancel()
	}