// Subscribe to a channel. Returns the total number of (p)subscriptions after
// subscribing.
516118
func (s *Subscriber) Subscribe(c string) int {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.channels[c] = struct{}{}
	return s.count()
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
192584
func (am *Manager) Subscribe(sink chan<- WalletEvent) event.Subscription {
	return am.feed.Subscribe(sink)
}
724536
func (s *Subscriber) Subscribe(e Event, f interface{}) *Subscriber {
	unsubscribe := s.events.subscribe(e, f)
	s.unsuscribes = append(s.unsuscribes, unsubscribe)
	return s
}
628157
func (s *SubscriptionProvider) Subscribe() *Subscription {
	s.Lock()
	defer s.Unlock()
	if s.subscriptions == nil {
		s.subscriptions = make(map[string]*Subscription)
	}
	sub := newSubscription(s)
	s.subscriptions[sub.id.String()] = sub

	return sub
}
871817
func (sub *Subscription) subUnsub(cmd string, channels ...string) error {
	err := sub.ensureProtocol()
	if err != nil {
		return err
	}
	pattern := false
	args := []interface{}{}
	for _, channel := range channels {
		if containsPattern(channel) {
			pattern = true
		}
		args = append(args, channel)
	}
	if pattern {
		cmd = "p" + cmd
	}
	err = sub.resp.sendCommand(cmd, args...)
	logCommand(cmd, args, err, sub.database.logging)
	return err
}
385886
func (cln *Client) Subscribe(c *ble.Characteristic, ind bool, fn ble.NotificationHandler) error {
	cln.conn.Lock()
	defer cln.conn.Unlock()
	cln.conn.subs[c.Handle] = &sub{fn: fn, char: c}
	rsp := cln.conn.sendReq(68, xpc.Dict{
		"kCBMsgArgDeviceUUID":                cln.id,
		"kCBMsgArgCharacteristicHandle":      c.Handle,
		"kCBMsgArgCharacteristicValueHandle": c.ValueHandle,
		"kCBMsgArgState":                     1,
	})
	if rsp.err() != nil {
		delete(cln.conn.subs, c.Handle)
		return rsp.err()
	}
	return nil
}
1086955
func (s *Subscriber) Channels() []string {
	s.mu.Lock()
	defer s.mu.Unlock()

	var cs []string
	for c := range s.channels {
		cs = append(cs, c)
	}
	sort.Strings(cs)
	return cs
}
1206742
func (p *Panel) Subscribe() <-chan PanelEvent {
	eventReader := make(chan PanelEvent)
	output := make(chan PanelEvent, 100)

	go func() {
		for v := range eventReader {
			select {
			case output <- v:
			default:
				<-output
				output <- v
			}
		}
	}()
	p.eventReceivers = append(p.eventReceivers, eventReader)
	return output
}
6464
func (t *peerThunk) Subscriber() peer.Subscriber {
	t.lock.RLock()
	s := t.subscriber
	t.lock.RUnlock()
	return s
}
258677
func (s *Service) Subscribe(topic string, qos packet.QOS) SubscribeFuture {
	return s.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
1177936
func (c *Client) subscribe() error {
	sub := c.protocol.Subscribe()
	events := sub.Events()

	go func() {
		for {
			select {
			case <-c.quitChan:
				return
			default:
			}
			select {
			case <-c.quitChan:
				return
			case event := <-events:
				switch event.(type) {
				case common.EventNewDevice,
					common.EventNewGroup,
					common.EventNewLocation,
					common.EventExpiredDevice,
					common.EventExpiredGroup,
					common.EventExpiredLocation:
					c.Notify(event)
				}
			}
		}
	}()

	return nil
}
1277829
func (h *Hub) Subscribe(kind Kind, f func(Event)) (cancel func()) {
	var cancelled bool
	h.m.Lock()
	h.seq++
	id := h.seq
	if h.subscribers == nil {
		h.subscribers = make(map[Kind][]handler)
	}
	h.subscribers[kind] = append(h.subscribers[kind], handler{id: id, f: f})
	h.m.Unlock()
	return func() {
		h.m.Lock()
		if cancelled {
			h.m.Unlock()
			return
		}
		cancelled = true
		a := h.subscribers[kind]
		for i, f := range a {
			if f.id == id {
				a[i], h.subscribers[kind] = a[len(a)-1], a[:len(a)-1]
				break
			}
		}
		if len(a) == 0 {
			delete(h.subscribers, kind)
		}
		h.m.Unlock()
	}
}
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()
}
1233451
func (c *Client) Subscribe(listener *Listener) {
	listener.Active = true
	c.tasks <- task{Listener: listener, Action: subscribeAction}
}
13472
func (e *Events) Subscribe() ([]eventtypes.Message, chan interface{}, func()) {
	eventSubscribers.Inc()
	e.mu.Lock()
	current := make([]eventtypes.Message, len(e.events))
	copy(current, e.events)
	l := e.pub.Subscribe()
	e.mu.Unlock()

	cancel := func() {
		e.Evict(l)
	}
	return current, l, cancel
}
661512
func NewSubscriptions(conn *Conn) *Subscriptions {
	s := &Subscriptions{
		channels:       make(map[string]struct{}),
		pchannels:      make(map[string]struct{}),
		conn:           conn,
		messageChannel: make(chan *Message, 100),
		ready:          true,
		readyChannel:   make(chan bool, 1),
		throwError:     false,
	}
	go s.receive()
	return s
}