// Subscribe creates an async subscription to receive notifications when the
// manager detects the arrival or departure of a wallet from any of its backends.
192584
func (am *Manager) Subscribe(sink chan<- WalletEvent) event.Subscription {
	return am.feed.Subscribe(sink)
}
1261518
func (e *EventListener) Subscribe(appID string, objectTypes []string, objectID string) (*EventSubscriber, error) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if e.IsClosed() {
		return nil, errors.New("event listener closed")
	}
	s := &EventSubscriber{
		Events:      make(chan *ct.Event),
		l:           e,
		queue:       make(chan *ct.Event, eventBufferSize),
		stop:        make(chan struct{}),
		appID:       appID,
		objectTypes: objectTypes,
		objectID:    objectID,
	}
	go s.loop()
	if _, ok := e.subscribers[appID]; !ok {
		e.subscribers[appID] = make(map[*EventSubscriber]struct{})
	}
	e.subscribers[appID][s] = struct{}{}
	return s, nil
}
1264673
func handleSubscribe(proxy *mist.Proxy, msg mist.Message) error {
	proxy.Subscribe(msg.Tags)
	return nil
}
1230982
func (a *Aggregator) Subscribe(id string, msgc chan<- *rfc5424.Message, donec <-chan struct{}) {
	a.getBuffer(id).Subscribe(msgc, donec)
}
724536
func (s *Subscriber) Subscribe(e Event, f interface{}) *Subscriber {
	unsubscribe := s.events.subscribe(e, f)
	s.unsuscribes = append(s.unsuscribes, unsubscribe)
	return s
}
687057
func (b *pubsubBroker) Subscribe(topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	options := broker.SubscribeOptions{
		AutoAck: true,
		Queue:   "q-" + uuid.New().String(),
		Context: b.options.Context,
	}

	for _, o := range opts {
		o(&options)
	}

	ctx := context.Background()
	sub := b.client.Subscription(options.Queue)

	if createSubscription, ok := b.options.Context.Value(createSubscription{}).(bool); !ok || createSubscription {
		exists, err := sub.Exists(ctx)
		if err != nil {
			return nil, err
		}

		if !exists {
			tt := b.client.Topic(topic)
			subb, err := b.client.CreateSubscription(ctx, options.Queue, pubsub.SubscriptionConfig{
				Topic:       tt,
				AckDeadline: time.Duration(0),
			})
			if err != nil {
				return nil, err
			}
			sub = subb
		}
	}

	subscriber := &subscriber{
		options: options,
		topic:   topic,
		exit:    make(chan bool),
		sub:     sub,
	}

	go subscriber.run(h)

	return subscriber, nil
}
1059862
func (p *PubSub) Subscribe(channels ...string) error {
	args := packArgs("SUBSCRIBE", channels)
	return p.conn.SendCommand(args...)
}
1180678
func (b *redisBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	var options broker.SubscribeOptions
	for _, o := range opts {
		o(&options)
	}

	s := subscriber{
		codec:  b.opts.Codec,
		conn:   &redis.PubSubConn{Conn: b.pool.Get()},
		topic:  topic,
		handle: handler,
		opts:   options,
	}

	// Run the receiver routine.
	go s.recv()

	if err := s.conn.Subscribe(s.topic); err != nil {
		return nil, err
	}

	return &s, nil
}
394916
func (p *Proxy) Subscribe(tags []string) {
	lumber.Trace("Proxy subscribing to '%s'...", tags)

	if len(tags) == 0 {
		return
	}

	// add proxy to subscribers list here so not all clients are 'subscribers'
	// since gets added to a map, there are no duplicates
	subscribe(p)

	// add tags to subscription
	p.Lock()
	p.subscriptions.Add(tags)
	p.Unlock()
}
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
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}
718356
func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool) (*Subscription, error) {
	if nc == nil {
		return nil, ErrInvalidConnection
	}
	nc.mu.Lock()
	// ok here, but defer is generally expensive
	defer nc.mu.Unlock()

	// Check for some error conditions.
	if nc.isClosed() {
		return nil, ErrConnectionClosed
	}
	if nc.isDraining() {
		return nil, ErrConnectionDraining
	}

	if cb == nil && ch == nil {
		return nil, ErrBadSubscription
	}

	sub := &Subscription{Subject: subj, Queue: queue, mcb: cb, conn: nc}
	// Set pending limits.
	sub.pMsgsLimit = DefaultSubPendingMsgsLimit
	sub.pBytesLimit = DefaultSubPendingBytesLimit

	// If we have an async callback, start up a sub specific
	// Go routine to deliver the messages.
	if cb != nil {
		sub.typ = AsyncSubscription
		sub.pCond = sync.NewCond(&sub.mu)
		go nc.waitForMsgs(sub)
	} else if !isSync {
		sub.typ = ChanSubscription
		sub.mch = ch
	} else { // Sync Subscription
		sub.typ = SyncSubscription
		sub.mch = ch
	}

	nc.subsMu.Lock()
	nc.ssid++
	sub.sid = nc.ssid
	nc.subs[sub.sid] = sub
	nc.subsMu.Unlock()

	// We will send these for all subs when we reconnect
	// so that we can suppress here if reconnecting.
	if !nc.isReconnecting() {
		fmt.Fprintf(nc.bw, subProto, subj, queue, sub.sid)
		// Kick flusher if needed.
		if len(nc.fch) == 0 {
			nc.kickFlusher()
		}
	}

	return sub, nil
}
1086425
func (am *Manager) update() {
	// Close all subscriptions when the manager terminates
	defer func() {
		am.lock.Lock()
		for _, sub := range am.updaters {
			sub.Unsubscribe()
		}
		am.updaters = nil
		am.lock.Unlock()
	}()

	// Loop until termination
	for {
		select {
		case event := <-am.updates:
			// Wallet event arrived, update local cache
			am.lock.Lock()
			switch event.Kind {
			case WalletArrived:
				am.wallets = merge(am.wallets, event.Wallet)
			case WalletDropped:
				am.wallets = drop(am.wallets, event.Wallet)
			}
			am.lock.Unlock()

			// Notify any listeners of the event
			am.feed.Send(event)

		case errc := <-am.quit:
			// Manager terminating, return
			errc <- nil
			return
		}
	}
}
1166201
func (c PubSubConn) Subscribe(channel ...interface{}) error {
	c.Conn.Send("SUBSCRIBE", channel...)
	return c.Conn.Flush()
}
1024250
func (em *EventMeter) Subscribe(query string, cb EventCallbackFunc) error {
	em.mtx.Lock()
	defer em.mtx.Unlock()

	if err := em.wsc.Subscribe(context.TODO(), query); err != nil {
		return err
	}

	metric := &EventMetric{
		meter:    metrics.NewMeter(),
		callback: cb,
	}
	em.queryToMetricMap[query] = metric
	return nil
}
512752
func (nc *Conn) SubscribeSync(subj string) (*Subscription, error) {
	if nc == nil {
		return nil, ErrInvalidConnection
	}
	mch := make(chan *Msg, nc.Opts.SubChanLen)
	s, e := nc.subscribe(subj, _EMPTY_, nil, mch, true)
	return s, e
}
611859
func (o *os) Subscribe(ctx context.Context, h Handler, types ...string) error {
	req := &ev.StreamRequest{}
	for _, typ := range types {
		req.Types = append(req.Types, typ)
	}

	stream, err := o.cl.Stream(ctx, req)
	if err != nil {
		return err
	}

	for {
		rsp, err := stream.Recv()
		if err != nil {
			return err
		}
		h(toRecord(rsp.Record))
	}

	return nil
}