// This is invoked when the first (or last) queue subscription on a
// given subject/group is registered (or unregistered). Sent to all
// inbound gateways.
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()
	}
}
612662
func (n *Notifier) takeSubscription() *Subscription {
	n.mu.Lock()
	defer n.mu.Unlock()
	n.callReturned = true
	return n.sub
}
258677
func (s *Service) Subscribe(topic string, qos packet.QOS) SubscribeFuture {
	return s.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
645725
func (cgi *ConsumergroupInstance) RegisterWithSubscription(subscriptionJSON []byte) error {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	err := cgi.cg.kz.create(node, subscriptionJSON, true)
	if err == zk.ErrNodeExists {
		return ErrInstanceAlreadyRegistered
	}
	return err
}
780948
func subscribe(p *Proxy) {
	lumber.Trace("Adding proxy to subscribers...")

	mutex.Lock()
	subscribers[p.id] = p
	mutex.Unlock()
}
414197
func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, queue, cb, nil, false)
}
860532
func (s *Server) broadcastUnSubscribe(sub *subscription) {
	trace := atomic.LoadInt32(&s.logging.trace) == 1
	s.mu.Lock()
	subs := []*subscription{sub}
	for _, route := range s.routes {
		route.mu.Lock()
		route.sendRouteUnSubProtos(subs, trace, func(sub *subscription) bool {
			return route.canImport(string(sub.subject))
		})
		route.mu.Unlock()
	}
	s.mu.Unlock()
}
1175957
func (o *CountdownAPI) RegisterConsumer(mediaType string, consumer runtime.Consumer) {
	o.customConsumers[mediaType] = consumer
}
30631
func (mr *MockAgentConnectorMockRecorder) QueueSubscribe(ctx, name, subject, group, output interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "QueueSubscribe", reflect.TypeOf((*MockAgentConnector)(nil).QueueSubscribe), ctx, name, subject, group, output)
}
655826
func (p *Pubsub) resubscribe(write *writePump) {
	timer := gaugeLatency(PromReconnectLatency)
	PromReconnections.Inc()

	p.subsMu.Lock()
	defer p.subsMu.Unlock()

	for kind, recs := range p.subs {
		if recs == nil {
			continue
		}

		for _, ev := range recs.getList() {
			write.Data() <- command{
				command: EventType(kind).SubCommand(),
				channel: ev.name,
			}
		}
	}

	timer()
}
1075340
func (tsm *TargetStatsMultiplexer) Subscribe() (int, <-chan (*TargetStatsEntry)) {
	i := tsm.nextIndex
	tsm.nextIndex++
	c := make(chan (*TargetStatsEntry), 100)
	tsm.listeners[i] = c
	return i, c
}
1316556
func (c *client) gatewayAllSubsReceiveStart(info *Info) {
	account := getAccountFromGatewayCommand(c, info, "start")
	if account == "" {
		return
	}
	// Since the remote would send us this start command
	// only after sending us too many RS- for this account,
	// we should always have an entry here.
	// TODO(ik): Should we close connection with protocol violation
	// error if that happens?
	ei, _ := c.gw.outsim.Load(account)
	if ei != nil {
		e := ei.(*outsie)
		e.Lock()
		e.mode = modeTransitioning
		e.Unlock()
	} else {
		e := &outsie{sl: NewSublist()}
		e.mode = modeTransitioning
		c.mu.Lock()
		c.gw.outsim.Store(account, e)
		c.mu.Unlock()
	}
}
1372572
func (c *client) processInboundRoutedMsg(msg []byte) {
	// Update statistics
	c.in.msgs++
	// The msg includes the CR_LF, so pull back out for accounting.
	c.in.bytes += int32(len(msg) - LEN_CR_LF)

	if c.trace {
		c.traceMsg(msg)
	}

	if c.opts.Verbose {
		c.sendOK()
	}

	// Mostly under testing scenarios.
	if c.srv == nil {
		return
	}

	acc, r := c.getAccAndResultFromCache()
	if acc == nil {
		c.Debugf("Unknown account %q for routed message on subject: %q", c.pa.account, c.pa.subject)
		return
	}

	// Check to see if we need to map/route to another account.
	if acc.imports.services != nil {
		c.checkForImportServices(acc, msg)
	}

	// Check for no interest, short circuit if so.
	// This is the fanout scale.
	if len(r.psubs)+len(r.qsubs) == 0 {
		return
	}

	// Check to see if we have a routed message with a service reply.
	if isServiceReply(c.pa.reply) && acc != nil {
		// Need to add a sub here for local interest to send a response back
		// to the originating server/requestor where it will be re-mapped.
		sid := make([]byte, 0, len(acc.Name)+len(c.pa.reply)+1)
		sid = append(sid, acc.Name...)
		sid = append(sid, ' ')
		sid = append(sid, c.pa.reply...)
		// Copy off the reply since otherwise we are referencing a buffer that will be reused.
		reply := make([]byte, len(c.pa.reply))
		copy(reply, c.pa.reply)
		sub := &subscription{client: c, subject: reply, sid: sid, max: 1}
		if err := acc.sl.Insert(sub); err != nil {
			c.Errorf("Could not insert subscription: %v", err)
		} else {
			ttl := acc.AutoExpireTTL()
			c.mu.Lock()
			c.subs[string(sid)] = sub
			c.addReplySubTimeout(acc, sub, ttl)
			c.mu.Unlock()
		}
	}
	c.processMsgResults(acc, r, msg, c.pa.subject, c.pa.reply, pmrNoFlag)
}
300615
func (conR *ConsensusReactor) subscribeToBroadcastEvents() {
	const subscriber = "consensus-reactor"
	conR.conS.evsw.AddListenerForEvent(subscriber, types.EventNewRoundStep,
		func(data tmevents.EventData) {
			conR.broadcastNewRoundStepMessage(data.(*cstypes.RoundState))
		})

	conR.conS.evsw.AddListenerForEvent(subscriber, types.EventValidBlock,
		func(data tmevents.EventData) {
			conR.broadcastNewValidBlockMessage(data.(*cstypes.RoundState))
		})

	conR.conS.evsw.AddListenerForEvent(subscriber, types.EventVote,
		func(data tmevents.EventData) {
			conR.broadcastHasVoteMessage(data.(*types.Vote))
		})

}
461832
func (s *Server) registerInboundGatewayConnection(cid uint64, gwc *client) {
	s.gateway.Lock()
	s.gateway.in[cid] = gwc
	s.gateway.Unlock()
}
489678
func (cgi *ConsumergroupInstance) Registered() (bool, error) {
	node := fmt.Sprintf("%s/consumers/%s/ids/%s", cgi.cg.kz.conf.Chroot, cgi.cg.Name, cgi.ID)
	return cgi.cg.kz.exists(node)
}
817262
func (conn *Connection) QueueSubscribe(ctx context.Context, name string, subject string, group string, output chan *ConnectorMessage) error {
	var err error

	s := &channelSubscription{
		in:   make(chan *nats.Msg, cap(output)),
		out:  output,
		quit: make(chan interface{}, 1),
	}

	conn.subMu.Lock()
	conn.chanSubscriptions[name] = s
	conn.subMu.Unlock()

	copier := func(ctx context.Context, s *channelSubscription) {
		for {
			select {
			case m := <-s.in:
				s.out <- &ConnectorMessage{Data: m.Data, Reply: m.Reply, Subject: m.Subject}
			case <-ctx.Done():
				conn.Unsubscribe(name)
				return
			case <-s.quit:
				close(s.in)
				return
			}
		}
	}

	go copier(ctx, s)

	conn.logger.Debugf("Subscribing to %s in group '%s' on server %s", subject, group, conn.ConnectedServer())

	s.subscription, err = conn.nats.ChanQueueSubscribe(subject, group, s.in)
	if err != nil {
		return fmt.Errorf("Could not subscribe to subscription %s: %s", name, err)
	}

	return err
}