// Subscribe watches etcd changes and generates structured events telling vulcand to add or delete frontends, hosts etc.
// It is a blocking function.
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
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
1334692
func (s *SubSocket) Subscribe(topic string) error {
	return s.sock.SetOption(mangos.OptionSubscribe, topic)
}
505065
func (p *ExamplePlugin) subscribeWatcher() (err error) {
	prefix := etcdKeyPrefix(p.ServiceLabel.GetAgentLabel())
	p.Log.Infof("Prefix: %v", prefix)

	p.watchDataReg, err = p.Watcher.Watch("ExamplePlugin",
		p.changeChannel, p.resyncChannel, prefix)
	if err != nil {
		return err
	}

	p.Log.Info("KeyValProtoWatcher subscribed")

	return 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()
}
1096281
func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Subscription {
	return bc.scope.Track(bc.chainSideFeed.Subscribe(ch))
}
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
}
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
}
520541
func (p *Peer) Subscribe(sub peer.Subscriber) {
	p.lock.Lock()
	p.subscribers[sub] = struct{}{}
	p.lock.Unlock()
}
592549
func (p *PubSub) Subscribe(id string, s Subscriber) error {
	p.m.Lock()
	defer p.m.Unlock()

	p.subscribers[id] = s
	return nil
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}
1049622
func (hub *Hub) Subscribe(sink chan<- accounts.WalletEvent) event.Subscription {
	// We need the mutex to reliably start/stop the update loop
	hub.stateLock.Lock()
	defer hub.stateLock.Unlock()

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

	// Subscribers require an active notification loop, start it
	if !hub.updating {
		hub.updating = true
		go hub.updater()
	}
	return sub
}
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
}
1136827
func (pm *Manager) SubscribeEvents(buffer int, watchEvents ...Event) (eventCh <-chan interface{}, cancel func()) {
	topic := func(i interface{}) bool {
		observed, ok := i.(Event)
		if !ok {
			panic(fmt.Sprintf("unexpected type passed to event channel: %v", reflect.TypeOf(i)))
		}
		for _, e := range watchEvents {
			if e.matches(observed) {
				return true
			}
		}
		// If no specific events are specified always assume a matched event
		// If some events were specified and none matched above, then the event
		// doesn't match
		return watchEvents == nil
	}
	ch := pm.publisher.SubscribeTopicWithBuffer(topic, buffer)
	cancelFunc := func() { pm.publisher.Evict(ch) }
	return ch, cancelFunc
}
96004
func (b *Buffer) subscribe(msgc chan<- *rfc5424.Message, donec <-chan struct{}) {
	b.subs[msgc] = struct{}{}

	go func() {
		select {
		case <-donec:
		case <-b.donec:
		}

		b.mu.Lock()
		defer b.mu.Unlock()

		delete(b.subs, msgc)
		close(msgc)
	}()
}
302057
func (c *EncodedConn) subscribe(subject, queue string, cb Handler) (*Subscription, error) {
	if cb == nil {
		return nil, errors.New("nats: Handler required for EncodedConn Subscription")
	}
	argType, numArgs := argInfo(cb)
	if argType == nil {
		return nil, errors.New("nats: Handler requires at least one argument")
	}

	cbValue := reflect.ValueOf(cb)
	wantsRaw := (argType == emptyMsgType)

	natsCB := func(m *Msg) {
		var oV []reflect.Value
		if wantsRaw {
			oV = []reflect.Value{reflect.ValueOf(m)}
		} else {
			var oPtr reflect.Value
			if argType.Kind() != reflect.Ptr {
				oPtr = reflect.New(argType)
			} else {
				oPtr = reflect.New(argType.Elem())
			}
			if err := c.Enc.Decode(m.Subject, m.Data, oPtr.Interface()); err != nil {
				if c.Conn.Opts.AsyncErrorCB != nil {
					c.Conn.ach.push(func() {
						c.Conn.Opts.AsyncErrorCB(c.Conn, m.Sub, errors.New("nats: Got an error trying to unmarshal: "+err.Error()))
					})
				}
				return
			}
			if argType.Kind() != reflect.Ptr {
				oPtr = reflect.Indirect(oPtr)
			}

			// Callback Arity
			switch numArgs {
			case 1:
				oV = []reflect.Value{oPtr}
			case 2:
				subV := reflect.ValueOf(m.Subject)
				oV = []reflect.Value{subV, oPtr}
			case 3:
				subV := reflect.ValueOf(m.Subject)
				replyV := reflect.ValueOf(m.Reply)
				oV = []reflect.Value{subV, replyV, oPtr}
			}

		}
		cbValue.Call(oV)
	}

	return c.Conn.subscribe(subject, queue, natsCB, nil, false)
}
517590
func (c *Client) EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (*ClientSubscription, error) {
	return c.Subscribe(ctx, "eth", channel, args...)
}