// Subscribe subscribes to an SQS queue, starting a goroutine to poll for messages
435108
func (b *sqsBroker) Subscribe(queueName string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	queueURL, err := b.urlFromQueueName(queueName)
	if err != nil {
		return nil, err
	}

	options := broker.SubscribeOptions{
		AutoAck: true,
		Queue:   queueName,
		Context: context.Background(),
	}

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

	subscriber := &subscriber{
		options:   options,
		URL:       queueURL,
		queueName: queueName,
		svc:       b.svc,
		exit:      make(chan bool),
	}
	go subscriber.run(h)

	return subscriber, nil
}
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()
		}
	}
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
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
}
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
}
1310917
func (s *Socket) Subscribe(filter []byte) {
	s.setBinary(C.ZMQ_SUBSCRIBE, filter)
}
371820
func (self *Client) Subscribe(channel string) (ch *Channel) {
	for _, ch := range self.Channels {
		if ch.Name == channel {
			self._subscribe <- ch
			return ch
		}
	}
	ch = &Channel{Name: channel, bindings: &self.bindings}
	self._subscribe <- ch
	return
}
109658
func (sc *conn) QueueSubscribe(subject, qgroup string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
	return sc.subscribe(subject, qgroup, cb, options...)
}
674167
func (m *MemoryBackend) Subscribe(client *Client, subs []packet.Subscription, ack Ack) error {
	// acquire global mutex
	m.globalMutex.Lock()
	defer m.globalMutex.Unlock()

	// save subscription
	for _, sub := range subs {
		client.Session().(*memorySession).subscriptions.Set(sub.Topic, sub)
	}

	// call ack if provided
	if ack != nil {
		ack()
	}

	// get session
	sess := client.Session().(*memorySession)

	// handle all subscriptions
	for _, sub := range subs {
		// get retained messages
		values := m.retainedMessages.Search(sub.Topic)

		// publish messages
		for _, value := range values {
			// add to temporary queue or return error if queue is full
			select {
			case sess.temporary <- value.(*packet.Message):
			default:
				return ErrQueueFull
			}
		}
	}

	return nil
}
204256
func (c *Consumer) Subscribe(topic string, rebalanceCb RebalanceCb) error {
	return c.SubscribeTopics([]string{topic}, rebalanceCb)
}
896960
func subscribe(sessions chan chan session, messages chan<- message) {
	queue := identity()

	for session := range sessions {
		sub := <-session

		if _, err := sub.QueueDeclare(queue, false, true, true, false, nil); err != nil {
			log.Printf("cannot consume from exclusive queue: %q, %v", queue, err)
			return
		}

		routingKey := "application specific routing key for fancy toplogies"
		if err := sub.QueueBind(queue, routingKey, exchange, false, nil); err != nil {
			log.Printf("cannot consume without a binding to exchange: %q, %v", exchange, err)
			return
		}

		deliveries, err := sub.Consume(queue, "", false, true, false, false, nil)
		if err != nil {
			log.Printf("cannot consume from: %q, %v", queue, err)
			return
		}

		log.Printf("subscribed...")

		for msg := range deliveries {
			messages <- message(msg.Body)
			sub.Ack(msg.DeliveryTag, false)
		}
	}
}
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
}
1019168
func (rcvr *Receiver) Subscribe(svcName string) {
	for _, subName := range rcvr.Subscriptions {
		if subName == svcName {
			return
		}
	}

	rcvr.Subscriptions = append(rcvr.Subscriptions, svcName)
}
724536
func (s *Subscriber) Subscribe(e Event, f interface{}) *Subscriber {
	unsubscribe := s.events.subscribe(e, f)
	s.unsuscribes = append(s.unsuscribes, unsubscribe)
	return s
}
1224240
func (n *Stan) Subscribe(route string, handler stan.MsgHandler, opts ...stan.SubscriptionOption) error {
	subsc, err := n.Con.Subscribe(route, handler, opts...)
	if err != nil {
		return errors.New("Failed to make subcriptions for " + route + ": " + err.Error())
	}
	n.Opts.routes = append(n.Opts.routes, &Route{route: route, handler: handler, subsc: subsc})
	return nil
}
476822
func (mgr *SettingsManager) Subscribe(subCh chan<- *ArgoCDSettings) {
	mgr.mutex.Lock()
	defer mgr.mutex.Unlock()
	mgr.subscribers = append(mgr.subscribers, subCh)
	log.Infof("%v subscribed to settings updates", subCh)
}
520541
func (p *Peer) Subscribe(sub peer.Subscriber) {
	p.lock.Lock()
	p.subscribers[sub] = struct{}{}
	p.lock.Unlock()
}