// Subscribe registers the EventHandler to be called for every message in the provided topic.
542885
func (c *Client) Subscribe(topic string, fn EventHandler) error {
	id := NewID()
	c.registerListener(id)
	sub := &Subscribe{
		Request: id,
		Options: make(map[string]interface{}),
		Topic:   URI(topic),
	}
	if err := c.Send(sub); err != nil {
		return err
	}
	// wait to receive SUBSCRIBED message
	msg, err := c.waitOnListener(id)
	if err != nil {
		return err
	} else if e, ok := msg.(*Error); ok {
		return fmt.Errorf("error subscribing to topic '%v': %v", topic, e.Error)
	} else if subscribed, ok := msg.(*Subscribed); !ok {
		return fmt.Errorf(formatUnexpectedMessage(msg, SUBSCRIBED))
	} else {
		// register the event handler with this subscription
		c.events[subscribed.Subscription] = &eventDesc{topic, fn}
	}
	return nil
}
179212
func (em *clientEventMonitor) Subscribe(ID string) (*Subscription, error) {
	em.Lock()
	defer em.Unlock()

	s := &Subscription{
		ID:            ID,
		cancelChannel: make(chan chan struct{}),
		eventChannel:  make(chan *dockerclient.APIEvents),
		monitorDone:   em.done,
		handlers:      make(map[string]HandlerFunc),
		monitor:       em,
		lock:          sync.RWMutex{},
	}

	em.subscriptions[ID] = append(em.subscriptions[ID], s)
	s.run()

	return s, nil
}
724536
func (s *Subscriber) Subscribe(e Event, f interface{}) *Subscriber {
	unsubscribe := s.events.subscribe(e, f)
	s.unsuscribes = append(s.unsuscribes, unsubscribe)
	return s
}
1264673
func handleSubscribe(proxy *mist.Proxy, msg mist.Message) error {
	proxy.Subscribe(msg.Tags)
	return nil
}
919496
func (r *marathonClient) Subscribe(callback string) error {
	path := fmt.Sprintf("%s?callbackUrl=%s", marathonAPISubscription, callback)
	return r.apiPost(path, "", nil)

}
2734
func (_m *DiagnosticsMessageSubscriber) Subscribe(_a0 appinsights.DiagnosticsMessageHandler) appinsights.DiagnosticsMessageListener {
	ret := _m.Called(_a0)

	var r0 appinsights.DiagnosticsMessageListener
	if rf, ok := ret.Get(0).(func(appinsights.DiagnosticsMessageHandler) appinsights.DiagnosticsMessageListener); ok {
		r0 = rf(_a0)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(appinsights.DiagnosticsMessageListener)
		}
	}

	return r0
}
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()
}
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
}
1226714
func (s *SubServer) Subscribe(serviceURL, sid string, timeoutInSeconds int, autoRefresh bool, sub Subscriber) (string, error) {
	req, err := http.NewRequest(
		"SUBSCRIBE",
		serviceURL,
		nil)

	if err != nil {
		return "", err
	}

	req.Header.Add("TIMEOUT", "Second-"+strconv.Itoa(timeoutInSeconds))
	req.Close = true

	if sid != "" {
		// Renew subscription
		req.Header.Add("SID", sid)
	} else {
		req.Header.Add("CALLBACK", "<http://"+s.listenAddr+">")
		req.Header.Add("NT", "upnp:event")
	}

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

	sids, ok := resp.Header["Sid"]
	if !ok || len(sids) == 0 {
		return "", errors.New("Sid header missing in response")
	}

	// If the sid == "" this is a new subscribe otherwise it is a refresh in
	// which case we don't need to insert this book keeping
	if sid == "" {
		s.subs[sids[0]] = &subInfo{
			SID:              sids[0],
			ServiceURL:       serviceURL,
			Subscriber:       sub,
			TimeoutInSeconds: timeoutInSeconds,
		}
	}

	if autoRefresh {
		s.renewRefreshTimer(sids[0])
	}

	return sids[0], nil
}
378282
func (c *DefaultClient) SubscribeAppEvents(appID string, eventType types.EventType, handler AppEventHandler) Token {
	topic := ApplicationTopic{appID, AppEvents, string(eventType)}
	return c.subscribe(topic.String(), func(mqtt MQTT.Client, msg MQTT.Message) {
		topic, err := ParseApplicationTopic(msg.Topic())
		if err != nil {
			c.ctx.Warnf("mqtt: received message on invalid events topic: %s", msg.Topic())
			return
		}
		handler(c, topic.AppID, types.EventType(topic.Field), msg.Payload())
	})
}
780948
func subscribe(p *Proxy) {
	lumber.Trace("Adding proxy to subscribers...")

	mutex.Lock()
	subscribers[p.id] = p
	mutex.Unlock()
}
143939
func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMessage) error {
	log.G(ctx).Debugf("Received subscription %s (selector: %v)", subscription.ID, subscription.Selector)

	publisher, cancel, err := w.publisherProvider.Publisher(ctx, subscription.ID)
	if err != nil {
		return err
	}
	// Send a close once we're done
	defer cancel()

	match := func(t *api.Task) bool {
		// TODO(aluzzardi): Consider using maps to limit the iterations.
		for _, tid := range subscription.Selector.TaskIDs {
			if t.ID == tid {
				return true
			}
		}

		for _, sid := range subscription.Selector.ServiceIDs {
			if t.ServiceID == sid {
				return true
			}
		}

		for _, nid := range subscription.Selector.NodeIDs {
			if t.NodeID == nid {
				return true
			}
		}

		return false
	}

	wg := sync.WaitGroup{}
	w.mu.Lock()
	for _, tm := range w.taskManagers {
		if match(tm.task) {
			wg.Add(1)
			go func(tm *taskManager) {
				defer wg.Done()
				tm.Logs(ctx, *subscription.Options, publisher)
			}(tm)
		}
	}
	w.mu.Unlock()

	// If follow mode is disabled, wait for the current set of matched tasks
	// to finish publishing logs, then close the subscription by returning.
	if subscription.Options == nil || !subscription.Options.Follow {
		waitCh := make(chan struct{})
		go func() {
			defer close(waitCh)
			wg.Wait()
		}()

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-waitCh:
			return nil
		}
	}

	// In follow mode, watch for new tasks. Don't close the subscription
	// until it's cancelled.
	ch, cancel := w.taskevents.Watch()
	defer cancel()
	for {
		select {
		case v := <-ch:
			task := v.(*api.Task)
			if match(task) {
				w.mu.RLock()
				tm, ok := w.taskManagers[task.ID]
				w.mu.RUnlock()
				if !ok {
					continue
				}

				go tm.Logs(ctx, *subscription.Options, publisher)
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}
1176840
func (c *ClusterClient) Subscribe(channels ...string) *PubSub {
	pubsub := c.pubSub()
	if len(channels) > 0 {
		_ = pubsub.Subscribe(channels...)
	}
	return pubsub
}
1271339
func (s *Store) Subscribe(service string, sendCurrent bool, kinds discoverd.EventKind, ch chan *discoverd.Event) stream.Stream {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Create service subscription list if it doesn't exist yet.
	if _, ok := s.subscribers[service]; !ok {
		s.subscribers[service] = list.New()
	}

	// Create and add subscription.
	sub := &subscription{
		kinds:   kinds,
		ch:      ch,
		store:   s,
		service: service,
	}
	sub.el = s.subscribers[service].PushBack(sub)

	// Send current instances.
	if sendCurrent && kinds.Any(discoverd.EventKindUp) {
		for _, inst := range s.instances(service) {
			ch <- &discoverd.Event{
				Service:  service,
				Kind:     discoverd.EventKindUp,
				Instance: inst,
			}
			// TODO: add a timeout to sends so that clients can't slow things down too much
		}
	}

	// Send current leader.
	if leader := s.serviceLeader(service); sendCurrent && kinds&discoverd.EventKindLeader != 0 && leader != nil {
		ch <- &discoverd.Event{
			Service:  service,
			Kind:     discoverd.EventKindLeader,
			Instance: leader,
		}
	}

	// Send current service meta data.
	if meta := s.serviceMeta(service); sendCurrent && kinds.Any(discoverd.EventKindServiceMeta) && meta != nil {
		ch <- &discoverd.Event{
			Service:     service,
			Kind:        discoverd.EventKindServiceMeta,
			ServiceMeta: meta,
		}
	}

	// Send current service.
	if sendCurrent && kinds.Any(discoverd.EventKindCurrent) {
		ch <- &discoverd.Event{
			Service: service,
			Kind:    discoverd.EventKindCurrent,
		}
	}

	return sub
}
1191218
func (p *Client) Subscribe(c *ble.Characteristic, ind bool, h ble.NotificationHandler) error {
	p.Lock()
	defer p.Unlock()
	if c.CCCD == nil {
		return fmt.Errorf("CCCD not found")
	}
	if ind {
		return p.setHandlers(c.CCCD.Handle, c.ValueHandle, cccIndicate, h)
	}
	return p.setHandlers(c.CCCD.Handle, c.ValueHandle, cccNotify, h)
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
789019
func (s *NatsSubscriber) Subscribe(subject, group string, handler nats.Handler) error {
	ch, err := s.server.initSubject(subject)
	if err != nil {
		return err
	}

	go func(s *NatsSubscriber, subject string, handler nats.Handler) {
		for r := range ch {
			handler.Process(&natsSubscription{subject: subject},
				&natsMessage{
					r: r,
				},
			)
		}
	}(s, subject, handler)
	return nil
}