// Subscribe implements Emitter.Subscribe
967012
func (p *Pubsub) Subscribe(ev EventBuilder, l Listener) {
	timer := gaugeLatency(PromSubLatency)
	defer timer()

	p.subsMu.Lock()
	count := p.subs[ev.kind].Add(ev, l)
	p.subsMu.Unlock()

	if count == 1 {
		PromSubscriptions.Inc()
		written := make(chan struct{}, 1)
		p.send <- command{
			command: ev.kind.SubCommand(),
			channel: ev.Name(),
			written: written,
		}

		<-written
	}
}
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
}
266196
func (s *SubscriptionSet) Subscribe() (<-chan map[string]*UnitStatus, <-chan error) {
	// TODO: Make fully evented by using systemd 209 with properties changed values
	return s.conn.SubscribeUnitsCustom(time.Second, 0,
		mismatchUnitStatus,
		func(unit string) bool { return s.filter(unit) },
	)
}
179204
func (kva *KvAlert) Subscribe(parentAlertType int64, childAlert *api.Alert) error {
	var subscriptions []api.Alert
	kv := kva.GetKvdbInstance()
	if _, err := kv.GetVal(getSubscriptionsKey(parentAlertType), &subscriptions); err != nil {
		if err != kvdb.ErrNotFound {
			return err
		}
	}
	subscriptions = append(subscriptions, *childAlert)
	_, err := kv.Put(getSubscriptionsKey(parentAlertType), subscriptions, 0)
	return err
}
952527
func (c *TCP) Subscribe(tags []string) error {

	if len(tags) == 0 {
		return fmt.Errorf("Unable to subscribe - missing tags")
	}

	return c.encoder.Encode(&mist.Message{Command: "subscribe", Tags: tags})
}
655153
func (whisper *Whisper) Subscribe(f *Filter) (string, error) {
	s, err := whisper.filters.Install(f)
	if err == nil {
		whisper.updateBloomFilter(f)
	}
	return s, err
}
1333766
func (p *pool) Subscribe(taskID string) {
	p.Lock()
	defer p.Unlock()

	if _, exists := p.subs[taskID]; !exists {
		// Version is the last item in the key, so we split here
		// to retrieve it for the subscription.
		p.subs[taskID] = &subscription{
			TaskID:  taskID,
			Version: p.version,
		}
	}
}
432539
func (n *ng) Subscribe(changes chan interface{}, afterIdx uint64, cancelC chan struct{}) error {
	w := n.kapi.Watcher(n.etcdKey, &etcd.WatcherOptions{AfterIndex: afterIdx, Recursive: true})
	for {
		response, err := w.Next(n.context)
		if err != nil {
			switch err {
			case context.Canceled:
				log.Infof("stop watching: graceful shutdown")
				return nil
			default:
				log.Errorf("unexpected error: %s, stop watching", err)
				return err
			}
		}
		log.Infof("%s", responseToString(response))
		change, err := n.parseChange(response)
		if err != nil {
			log.WithError(err).Warningf("ignore '%s'", responseToString(response))
			continue
		}
		if change != nil {
			log.Infof("%v", change)
			select {
			case changes <- change:
			case <-cancelC:
				return nil
			}
		}
	}
}
328861
func (c *pubsubClient) Subscribe(topic string, otherTopics ...string) (err Error) {
	args := appendAndConvert(topic, otherTopics...)
	//	var ok bool
	_, err = c.conn.ServiceRequest(&SUBSCRIBE, args)
	//	if err == nil {
	//		err = NewError(REDIS_ERR, "Subscribe() NOT IMPLEMENTED")
	//	}
	return
}
731710
func (_m *Light) Subscribe() *common.Subscription {
	ret := _m.Called()

	var r0 *common.Subscription
	if rf, ok := ret.Get(0).(func() *common.Subscription); ok {
		r0 = rf()
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*common.Subscription)
		}
	}

	return r0
}
260127
func (c *WSClient) Subscribe(ctx context.Context, query string) error {
	params := map[string]interface{}{"query": query}
	return c.Call(ctx, "subscribe", params)
}
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
}
919496
func (r *marathonClient) Subscribe(callback string) error {
	path := fmt.Sprintf("%s?callbackUrl=%s", marathonAPISubscription, callback)
	return r.apiPost(path, "", nil)

}
528232
func (ks *KeyStore) Subscribe(sink chan<- accounts.WalletEvent) event.Subscription {
	// We need the mutex to reliably start/stop the update loop
	ks.mu.Lock()
	defer ks.mu.Unlock()

	// Subscribe the caller and track the subscriber count
	sub := ks.updateScope.Track(ks.updateFeed.Subscribe(sink))

	// Subscribers require an active notification loop, start it
	if !ks.updating {
		ks.updating = true
		go ks.updater()
	}
	return sub
}
1334692
func (s *SubSocket) Subscribe(topic string) error {
	return s.sock.SetOption(mangos.OptionSubscribe, topic)
}
325372
func (n *ng) Subscribe(changes chan interface{}, afterIdx uint64, cancelC chan struct{}) error {
	watcher := etcd.NewWatcher(n.client)
	defer watcher.Close()

	var watchChan etcd.WatchChan
	ready := make(chan struct{})
	go func() {
		watchChan = watcher.Watch(etcd.WithRequireLeader(n.context), n.etcdKey,
			etcd.WithRev(int64(afterIdx)), etcd.WithPrefix())
		close(ready)
	}()
	select {
	case <-ready:
		log.Infof("begin watching: etcd revision %d", afterIdx)
	case <-time.After(time.Second * 10):
		return errors.New("timed out while waiting for watcher.Watch() to start")
	}

	for response := range watchChan {
		if response.Canceled {
			log.Infof("stop watching: graceful shutdown")
			return nil
		}
		if err := response.Err(); err != nil {
			log.Errorf("stop watching: error: %v", err)
			return err
		}

		for _, event := range response.Events {
			log.Infof("%s", eventToString(event))
			change, err := n.parseChange(event)
			if err != nil {
				log.Warningf("ignore '%s', error: %s", eventToString(event), err)
				continue
			}
			if change != nil {
				log.Infof("%v", change)
				select {
				case changes <- change:
				case <-cancelC:
					return nil
				}
			}
		}
	}

	return nil
}
204256
func (c *Consumer) Subscribe(topic string, rebalanceCb RebalanceCb) error {
	return c.SubscribeTopics([]string{topic}, rebalanceCb)
}