// Subscription returns the current subscription as set by Subscribe()
430818
func (c *Consumer) Subscription() (topics []string, err error) {
	var cTopics *C.rd_kafka_topic_partition_list_t

	cErr := C.rd_kafka_subscription(c.handle.rk, &cTopics)
	if cErr != C.RD_KAFKA_RESP_ERR_NO_ERROR {
		return nil, newError(cErr)
	}
	defer C.rd_kafka_topic_partition_list_destroy(cTopics)

	topicCnt := int(cTopics.cnt)
	topics = make([]string, topicCnt)
	for i := 0; i < topicCnt; i++ {
		crktpar := C._c_rdkafka_topic_partition_list_entry(cTopics,
			C.int(i))
		topics[i] = C.GoString(crktpar.topic)
	}

	return topics, nil
}
720514
func (c *Client) Subscribe(name string) error {
	if err := c.ensureAuthenticated(); err != nil {
		return err
	}

	cmd := &commands.Subscribe{
		Mailbox: name,
	}

	status, err := c.execute(cmd, nil)
	if err != nil {
		return err
	}
	return status.Err()
}
1322813
func (em *EventMeter) subscribe() error {
	for query := range em.queryToMetricMap {
		if err := em.wsc.Subscribe(context.TODO(), query); err != nil {
			return err
		}
	}
	return nil
}
1344323
func (b *EventBus) subscription(m eh.EventMatcher, h eh.EventHandler, observer bool) *pubsub.Subscription {
	b.registeredMu.Lock()
	defer b.registeredMu.Unlock()

	if m == nil {
		panic("matcher can't be nil")
	}
	if h == nil {
		panic("handler can't be nil")
	}
	if _, ok := b.registered[h.HandlerType()]; ok {
		panic(fmt.Sprintf("multiple registrations for %s", h.HandlerType()))
	}
	b.registered[h.HandlerType()] = struct{}{}

	id := string(h.HandlerType())
	if observer { // Generate unique ID for each observer.
		id = fmt.Sprintf("%s-%s", id, uuid.New())
	}

	ctx := context.Background()

	// Get or create the subscription.
	subscriptionID := b.appID + "_" + id
	sub := b.client.Subscription(subscriptionID)
	if ok, err := sub.Exists(ctx); err != nil {
		panic("could not check subscription: " + err.Error())
	} else if !ok {
		if sub, err = b.client.CreateSubscription(ctx, subscriptionID,
			pubsub.SubscriptionConfig{
				Topic:       b.topic,
				AckDeadline: 60 * time.Second,
			},
		); err != nil {
			panic("could not create subscription: " + err.Error())
		}
	}

	return sub
}
204256
func (c *Consumer) Subscribe(topic string, rebalanceCb RebalanceCb) error {
	return c.SubscribeTopics([]string{topic}, rebalanceCb)
}
260127
func (c *WSClient) Subscribe(ctx context.Context, query string) error {
	params := map[string]interface{}{"query": query}
	return c.Call(ctx, "subscribe", params)
}
1310917
func (s *Socket) Subscribe(filter []byte) {
	s.setBinary(C.ZMQ_SUBSCRIBE, filter)
}
236720
func (w *wrapper) Subscribe(c Committer) {
	w.mut.Lock()
	w.subs = append(w.subs, c)
	w.mut.Unlock()
}
247956
func (br *defaultBroker) Subscribe(sub Sender, msg *Subscribe) {
	if _, ok := br.routes[msg.Topic]; !ok {
		br.routes[msg.Topic] = make(map[ID]Sender)
	}
	id := NewID()
	br.routes[msg.Topic][id] = sub

	br.subscriptions[id] = msg.Topic

	sub.Send(&Subscribed{Request: msg.Request, Subscription: id})
}
255359
func (sc *conn) subscribe(subject, qgroup string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
	sub := &subscription{subject: subject, qgroup: qgroup, inbox: nats.NewInbox(), cb: cb, sc: sc, opts: DefaultSubscriptionOptions}
	for _, opt := range options {
		if err := opt(&sub.opts); err != nil {
			return nil, err
		}
	}
	sc.Lock()
	if sc.nc == nil {
		sc.Unlock()
		return nil, ErrConnectionClosed
	}

	// Register subscription.
	sc.subMap[sub.inbox] = sub
	nc := sc.nc
	sc.Unlock()

	// Hold lock throughout.
	sub.Lock()
	defer sub.Unlock()

	// Listen for actual messages.
	nsub, err := nc.Subscribe(sub.inbox, sc.processMsg)
	if err != nil {
		return nil, err
	}
	sub.inboxSub = nsub

	// Create a subscription request
	// FIXME(dlc) add others.
	sr := &pb.SubscriptionRequest{
		ClientID:      sc.clientID,
		Subject:       subject,
		QGroup:        qgroup,
		Inbox:         sub.inbox,
		MaxInFlight:   int32(sub.opts.MaxInflight),
		AckWaitInSecs: int32(sub.opts.AckWait / time.Second),
		StartPosition: sub.opts.StartAt,
		DurableName:   sub.opts.DurableName,
	}

	// Conditionals
	switch sr.StartPosition {
	case pb.StartPosition_TimeDeltaStart:
		sr.StartTimeDelta = time.Now().UnixNano() - sub.opts.StartTime.UnixNano()
	case pb.StartPosition_SequenceStart:
		sr.StartSequence = sub.opts.StartSequence
	}

	b, _ := sr.Marshal()
	reply, err := nc.Request(sc.subRequests, b, sc.opts.ConnectTimeout)
	if err != nil {
		sub.inboxSub.Unsubscribe()
		if err == nats.ErrTimeout {
			err = ErrSubReqTimeout
		}
		return nil, err
	}
	r := &pb.SubscriptionResponse{}
	if err := r.Unmarshal(reply.Data); err != nil {
		sub.inboxSub.Unsubscribe()
		return nil, err
	}
	if r.Error != "" {
		sub.inboxSub.Unsubscribe()
		return nil, errors.New(r.Error)
	}
	sub.ackInbox = r.AckInbox

	return sub, nil
}
933546
func (c *Client) Subscribe(ctx context.Context, req *SubscriptionRequest) (string, error) {
	c.subscriptions.add(req)
	err := c.asynchronous.Send(ctx, req)
	if err != nil {
		// propagate send error
		return "", err
	}
	return req.SubID, nil
}
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
			}
		}
	}
}
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
}
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)
		}
	}
}
657656
func (m *Miniredis) NewSubscriber() *Subscriber {
	sub := newSubscriber()

	m.Lock()
	m.addSubscriber(sub)
	m.Unlock()

	return sub
}
1065184
func (cb *Breaker) Subscribe() <-chan BreakerEvent {
	eventReader := make(chan BreakerEvent)
	output := make(chan BreakerEvent, 100)

	go func() {
		for v := range eventReader {
			select {
			case output <- v:
			default:
				<-output
				output <- v
			}
		}
	}()
	cb.eventReceivers = append(cb.eventReceivers, eventReader)
	return output
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}