// Sends all subscriptions for the given account to the remove gateway
// This is sent from the inbound side, that is, the side that receives
// messages from the remote's outbound connection. This side is
// the one sending the subscription interest.
195289
func (s *Server) sendAccountSubsToGateway(c *client, accName []byte) {
	s.sendSubsToGateway(c, accName)
}
489623
func endSubscriber(m *Miniredis, c *server.Peer) {
	ctx := getCtx(c)
	if sub := ctx.subscriber; sub != nil {
		m.removeSubscriber(sub) // will Close() the sub
	}
	ctx.subscriber = nil
}
954756
func (c *client) sendAccountUnsubToGateway(accName []byte) {
	// Check if we have sent the A- or not.
	c.mu.Lock()
	e, sent := c.gw.insim[string(accName)]
	if e != nil || !sent {
		// Add a nil value to indicate that we have sent an A-
		// so that we know to send A+ when needed.
		c.gw.insim[string(accName)] = nil
		var protoa [256]byte
		proto := protoa[:0]
		proto = append(proto, aUnsubBytes...)
		proto = append(proto, accName...)
		proto = append(proto, CR_LF...)
		c.sendProto(proto, false)
		if c.trace {
			c.traceOutOp("", proto[:len(proto)-LEN_CR_LF])
		}
	}
	c.mu.Unlock()
}
840287
func (sc *SubscriptionScope) Close() {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	if sc.closed {
		return
	}
	sc.closed = true
	for s := range sc.subs {
		s.s.Unsubscribe()
	}
	sc.subs = nil
}
1163308
func (s *SubServer) Unsubscribe(sid string) error {
	sub, ok := s.subs[sid]

	if !ok {
		return nil
	}

	if sub.Timer != nil {
		sub.Timer.Stop()
	}

	delete(s.subs, sid)

	req, err := http.NewRequest(
		"UNSUBSCRIBE",
		sub.ServiceURL,
		nil)
	req.Close = true
	req.Header.Add("SID", sub.SID)

	client := &http.Client{}
	resp, err := client.Do(req)
	if resp != nil {
		defer resp.Body.Close()
	}

	if err != nil {
		return err
	}
	if resp.StatusCode != 200 {
		return fmt.Errorf("error unsubscribing, status code: %d", resp.StatusCode)
	}
	return nil
}
298292
func (c *client) processGatewayAccountSub(accName string) error {
	// Just to indicate activity around "subscriptions" events.
	c.in.subs++
	c.gw.outsim.Delete(accName)
	return nil
}
485266
func (d *IdentifyDispatch) Subscribe(mctx MetaContext) (unsubscribe func(), recvCh <-chan IdentifyDispatchMsg) {
	mctx.Debug("IdentifyDispatch.Subscribe")
	ch := make(chan IdentifyDispatchMsg, 10)
	d.Lock()
	defer d.Unlock()
	d.listeners = append(d.listeners, ch)
	unsubscribe = func() {
		mctx.Debug("IdentifyDispatch.Unsubcribe")
		d.Lock()
		defer d.Unlock()
		var listeners []chan<- IdentifyDispatchMsg
		for _, ch2 := range d.listeners {
			if ch == ch2 {
				continue
			}
			listeners = append(listeners, ch2)
		}
		d.listeners = listeners
	}
	return unsubscribe, ch
}
545987
func (s *Server) sendQueueSubOrUnsubToGateways(accName string, qsub *subscription, added bool) {
	if qsub.queue == nil {
		return
	}

	gwsa := [16]*client{}
	gws := gwsa[:0]
	s.getInboundGatewayConnections(&gws)
	if len(gws) == 0 {
		return
	}

	var protoa [512]byte
	var proto []byte
	for _, c := range gws {
		if proto == nil {
			proto = protoa[:0]
			if added {
				proto = append(proto, rSubBytes...)
			} else {
				proto = append(proto, rUnsubBytes...)
			}
			proto = append(proto, accName...)
			proto = append(proto, ' ')
			proto = append(proto, qsub.subject...)
			proto = append(proto, ' ')
			proto = append(proto, qsub.queue...)
			if added {
				// For now, just use 1 for the weight
				proto = append(proto, ' ', '1')
			}
			proto = append(proto, CR_LF...)
		}
		c.mu.Lock()
		c.sendProto(proto, false)
		if c.trace {
			c.traceOutOp("", proto[:len(proto)-LEN_CR_LF])
		}
		c.mu.Unlock()
	}
}
1205649
func (s *ListsService) Subscriptions(params *ListsSubscriptionsParams) (*Subscribed, *http.Response, error) {
	subscribed := new(Subscribed)
	apiError := new(APIError)
	resp, err := s.sling.New().Get("subscriptions.json").QueryStruct(params).Receive(subscribed, apiError)
	return subscribed, resp, relevantError(err, *apiError)
}
1126104
func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error {
	s.mtx.RLock()
	clientSubscriptions, ok := s.subscriptions[clientID]
	if ok {
		_, ok = clientSubscriptions[query.String()]
	}
	s.mtx.RUnlock()
	if !ok {
		return ErrSubscriptionNotFound
	}

	select {
	case s.cmds <- cmd{op: unsub, clientID: clientID, query: query}:
		s.mtx.Lock()
		delete(clientSubscriptions, query.String())
		if len(clientSubscriptions) == 0 {
			delete(s.subscriptions, clientID)
		}
		s.mtx.Unlock()
		return nil
	case <-ctx.Done():
		return ctx.Err()
	case <-s.Quit():
		return nil
	}
}
408345
func (bc *brokerConsumer) subscriptionConsumer() {
	<-bc.wait // wait for our first piece of work

	for newSubscriptions := range bc.newSubscriptions {
		bc.updateSubscriptions(newSubscriptions)

		if len(bc.subscriptions) == 0 {
			// We're about to be shut down or we're about to receive more subscriptions.
			// Either way, the signal just hasn't propagated to our goroutine yet.
			<-bc.wait
			continue
		}

		response, err := bc.fetchNewMessages()

		if err != nil {
			Logger.Printf("consumer/broker/%d disconnecting due to error processing FetchRequest: %s\n", bc.broker.ID(), err)
			bc.abort(err)
			return
		}

		bc.acks.Add(len(bc.subscriptions))
		for child := range bc.subscriptions {
			child.feeder <- response
		}
		bc.acks.Wait()
		bc.handleResponses()
	}
}
788200
func (s *Server) sendSubsToGateway(c *client, accountName []byte) {
	var (
		bufa = [32 * 1024]byte{}
		buf  = bufa[:0]
		epa  = [1024]int{}
		ep   = epa[:0]
	)

	gw := s.gateway

	// This needs to run under this lock for the whole duration
	gw.pasi.Lock()
	defer gw.pasi.Unlock()

	// Build the protocols
	buildProto := func(accName []byte, acc map[string]*sitally, doQueues bool) {
		for saq, si := range acc {
			if doQueues && si.q || !doQueues && !si.q {
				buf = append(buf, rSubBytes...)
				buf = append(buf, accName...)
				buf = append(buf, ' ')
				// For queue subs (si.q is true), saq will be
				// subject + ' ' + queue, for plain subs, this is
				// just the subject.
				buf = append(buf, saq...)
				if doQueues {
					buf = append(buf, ' ', '1')
				}
				buf = append(buf, CR_LF...)
				ep = append(ep, len(buf))
			}
		}
	}
	// If account is specified...
	if accountName != nil {
		// Simply send all plain subs (no queues) for this specific account
		buildProto(accountName, gw.pasi.m[string(accountName)], false)
		// Instruct to send all subs (RS+/-) for this account from now on.
		c.mu.Lock()
		e := c.gw.insim[string(accountName)]
		if e == nil {
			e = &insie{}
			c.gw.insim[string(accountName)] = e
		}
		e.mode = modeInterestOnly
		c.mu.Unlock()
	} else {
		// Send queues for all accounts
		for accName, acc := range gw.pasi.m {
			buildProto([]byte(accName), acc, true)
		}
	}

	// Nothing to send.
	if len(buf) == 0 {
		return
	}
	if len(buf) > cap(bufa) {
		s.Debugf("Sending subscriptions to %q, buffer size: %v", c.gw.name, len(buf))
	}
	// Send
	mbs := gw.sqbsz
	mp := int(c.out.mp / 2)
	if mbs > mp {
		mbs = mp
	}
	le := 0
	li := 0
	for i := 0; i < len(ep); i++ {
		if ep[i]-le > mbs {
			var end int
			// If single proto is bigger than our max buffer size,
			// send anyway. If it reaches a max in queueOutbound,
			// that function will close the connection.
			if i-li <= 1 {
				end = ep[i]
			} else {
				end = ep[i-1]
				i--
			}
			c.mu.Lock()
			c.queueOutbound(buf[le:end])
			c.flushOutbound()
			closed := c.flags.isSet(clearConnection)
			c.mu.Unlock()
			if closed {
				return
			}
			le = ep[i]
			li = i
		}
	}
	c.mu.Lock()
	c.queueOutbound(buf[le:])
	c.flushOutbound()
	if !c.flags.isSet(clearConnection) {
		c.Debugf("Sent queue subscriptions to gateway")
	}
	c.mu.Unlock()
}
133952
func (c *WSClient) UnsubscribeAll(ctx context.Context) error {
	params := map[string]interface{}{}
	return c.Call(ctx, "unsubscribe_all", params)
}
1144668
func (s *Server) removeRemoteGatewayConnection(c *client) {
	c.mu.Lock()
	cid := c.cid
	isOutbound := c.gw.outbound
	gwName := c.gw.name
	c.mu.Unlock()

	gw := s.gateway
	gw.Lock()
	if isOutbound {
		delete(gw.out, gwName)
		louto := len(gw.outo)
		reorder := false
		for i := 0; i < len(gw.outo); i++ {
			if gw.outo[i] == c {
				// If last, simply remove and no need to reorder
				if i != louto-1 {
					gw.outo[i] = gw.outo[louto-1]
					reorder = true
				}
				gw.outo = gw.outo[:louto-1]
			}
		}
		if reorder {
			gw.orderOutboundConnectionsLocked()
		}
	} else {
		delete(gw.in, cid)
	}
	gw.Unlock()
	s.removeFromTempClients(cid)

	if isOutbound {
		// Update number of totalQSubs for this gateway
		qSubsRemoved := int64(0)
		c.mu.Lock()
		for _, sub := range c.subs {
			if sub.queue != nil {
				qSubsRemoved++
			}
		}
		c.mu.Unlock()
		// Update total count of qsubs in remote gateways.
		atomic.AddInt64(&c.srv.gateway.totalQSubs, -qSubsRemoved)

	} else {
		var subsa [1024]*subscription
		var subs = subsa[:0]

		// For inbound GW connection, if we have subs, those are
		// local subs on "_R_." subjects.
		c.mu.Lock()
		for _, sub := range c.subs {
			subs = append(subs, sub)
		}
		c.mu.Unlock()
		for _, sub := range subs {
			c.removeReplySub(sub)
		}
	}
}
100337
func (c *client) sendAllAccountSubs() {
	// Hold all at once for now.
	var b bytes.Buffer

	c.mu.Lock()
	for key, n := range c.leaf.smap {
		c.writeLeafSub(&b, key, n)
	}

	// We will make sure we don't overflow here due to an max_pending.
	chunks := protoChunks(b.Bytes(), MAX_PAYLOAD_SIZE)
	for _, chunk := range chunks {
		c.queueOutbound(chunk)
		c.flushOutbound()
	}
	c.mu.Unlock()
}
1260128
func (sub *Subscription) Unsubscribe() {
	sub.unsubOnce.Do(func() {
	uninstallLoop:
		for {
			// write uninstall request and consume logs/hashes. This prevents
			// the eventLoop broadcast method to deadlock when writing to the
			// filter event channel while the subscription loop is waiting for
			// this method to return (and thus not reading these events).
			select {
			case sub.es.uninstall <- sub.f:
				break uninstallLoop
			case <-sub.f.logs:
			case <-sub.f.hashes:
			case <-sub.f.headers:
			}
		}

		// wait for filter to be uninstalled in work loop before returning
		// this ensures that the manager won't use the event channel which
		// will probably be closed by the client asap after this method returns.
		<-sub.Err()
	})
}
197188
func (s *Subscription) Drain() error {
	if s == nil {
		return ErrBadSubscription
	}
	s.mu.Lock()
	conn := s.conn
	s.mu.Unlock()
	if conn == nil {
		return ErrBadSubscription
	}
	return conn.unsubscribe(s, 0, true)
}