//  The subscriber thread requests messages starting with
//  A and B, then reads and counts incoming messages.
341336
func subscriber_thread() {
	//  Subscribe to "A" and "B"
	subscriber, _ := zmq.NewSocket(zmq.SUB)
	subscriber.Connect("tcp://localhost:6001")
	subscriber.SetSubscribe("A")
	subscriber.SetSubscribe("B")
	defer subscriber.Close() // cancel subscribe

	for count := 0; count < 5; count++ {
		_, err := subscriber.RecvMessage(0)
		if err != nil {
			break //  Interrupted
		}
	}
}
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)
}
257627
func (c *Client) Subscribe(channel string, m chan<- PubSubMessage, stop <-chan bool) error {
	srv, err := c.selector.PickServer("")
	if err != nil {
		return err
	}
	cn, err := c.getConn(srv)
	if err != nil {
		return err
	}
	// we cannot return this connection to the pool
	// because it'll be in subscribe context.
	//defer cn.condRelease(&err)
	_, err = c.execute(cn.rw, "SUBSCRIBE", channel)
	if err != nil {
		return err
	}
	if err = cn.nc.SetDeadline(time.Time{}); err != nil {
		cn.nc.Close()
		return err
	}
	watcher := make(chan struct{})
	go func() {
		select {
		case <-stop:
			cn.nc.Close()
		case <-watcher:
		}
	}()
	go func() {
		defer cn.nc.Close()
		for {
			raw, err := c.parseResponse(cn.rw.Reader)
			if err != nil {
				m <- PubSubMessage{
					Error: err,
				}
				close(watcher)
				return
			}
			switch raw.(type) {
			case []interface{}:
				ret := raw.([]interface{})
				m <- PubSubMessage{
					Value:   ret[2].(string),
					Channel: ret[1].(string),
					Error:   nil,
				}
			default:
				m <- PubSubMessage{
					Error: ErrServerError,
				}
				close(watcher)
				return
			}
		}
	}()
	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})
}
258493
func (c *EncodedConn) Subscribe(subject string, cb Handler) (*Subscription, error) {
	return c.subscribe(subject, _EMPTY_, cb)
}
71340
func (conn *ProtoConnection) ConsumeTopic(msgClb func(messaging.ProtoMessage), topics ...string) error {
	conn.multiplexer.rwlock.Lock()
	defer conn.multiplexer.rwlock.Unlock()

	if conn.multiplexer.started {
		return fmt.Errorf("ConsumeTopic can be called only if the multiplexer has not been started yet")
	}

	byteClb := func(bm *client.ConsumerMessage) {
		pm := client.NewProtoConsumerMessage(bm, conn.serializer)
		msgClb(pm)
	}

	for _, topic := range topics {
		// check if we have already consumed the topic
		var found bool
		var subs *consumerSubscription
	LoopSubs:
		for _, subscription := range conn.multiplexer.mapping {
			if subscription.manual == true {
				// do not mix dynamic and manual mode
				continue
			}
			if subscription.topic == topic {
				found = true
				subs = subscription
				break LoopSubs
			}
		}

		if !found {
			subs = &consumerSubscription{
				manual:         false, // non-manual example
				topic:          topic,
				connectionName: conn.name,
				byteConsMsg:    byteClb,
			}
			// subscribe new topic
			conn.multiplexer.mapping = append(conn.multiplexer.mapping, subs)
		}

		// add subscription to consumerList
		subs.byteConsMsg = byteClb
	}

	return nil
}
764339
func (s *ActivityService) GetThreadSubscription(ctx context.Context, id string) (*Subscription, *Response, error) {
	u := fmt.Sprintf("notifications/threads/%v/subscription", id)

	req, err := s.client.NewRequest("GET", u, nil)
	if err != nil {
		return nil, nil, err
	}

	sub := new(Subscription)
	resp, err := s.client.Do(ctx, req, sub)
	if err != nil {
		return nil, resp, err
	}

	return sub, resp, nil
}
1132399
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		start := time.Now()
		err := next(ctx, m)
		elapsed := time.Now().Sub(start)

		var traceID string
		span := opentracing.SpanFromContext(ctx)
		if span != nil {
			zs, ok := span.Context().(zipkintracing.SpanContext)
			if ok {
				traceID = zs.TraceID.ToHex()
			}
		}

		fields := logrus.Fields{
			"component": "pubsub",
			"topic":     opts.Topic,
			"handler":   opts.Name,
			"duration":  elapsed,
			"metadata":  m.Metadata,
		}

		if err != nil {
			de, ok := err.(errors.DropboxError)
			if ok {
				fields["err"] = de.GetMessage()
			} else {
				fields["err"] = err
			}
		}

		if m.ID != "" {
			fields["id"] = m.ID
		}

		if traceID != "" {
			fields["trace-id"] = traceID
		}

		logrus.WithFields(fields).Debug("Processed PubSub Msg")
		return err
	}
}
1166201
func (c PubSubConn) Subscribe(channel ...interface{}) error {
	c.Conn.Send("SUBSCRIBE", channel...)
	return c.Conn.Flush()
}
93413
func (p *Provider) Subscribe(opts pubsub.HandlerOptions, h pubsub.MsgHandler) {
	logrus.Infof("Subscribing to %s successWriter/name %s", opts.Topic, opts.ServiceName+"."+opts.Name)
	logSuccess := logrus.WithField("pubsub", "kafka")
	successWriter := logSuccess.Writer()
	p.Reader = kafka.NewReader(kafka.ReaderConfig{
		Brokers:        p.Brokers,
		GroupID:        opts.ServiceName + "." + opts.Name,
		Topic:          opts.Topic,
		MaxBytes:       10e6, // 10MB
		CommitInterval: time.Second,
		MaxWait:        50 * time.Millisecond,
		Logger:         log.New(successWriter, "", log.Lshortfile),
		ErrorLogger:    log.New(ioutil.Discard, "", log.Lshortfile),
	})

	b := &backoff.Backoff{
		Min:    200 * time.Millisecond,
		Max:    600 * time.Second,
		Factor: 2,
		Jitter: true,
	}

	go func() {
		for {
			ctx := context.Background()
			m, err := p.Reader.FetchMessage(ctx)
			if err != nil {
				d := b.Duration()
				logrus.Errorf(
					"Subscription receive to topic %s failed, reconnecting in %v. Err: %v",
					opts.Topic, d, err,
				)
				time.Sleep(d)
			}

			b.Reset()

			msg := pubsub.Msg{
				ID:   string(m.Key),
				Data: m.Value,
				Ack: func() {
					p.Reader.CommitMessages(ctx, m)
				},
				Nack: func() {},
			}

			err = h(ctx, msg)
			if err != nil {
				break
			}

			if opts.AutoAck {
				msg.Ack()
			}

			logrus.Debugf("message at topic/partition/offset %v/%v/%v\n",
				m.Topic, m.Partition, m.Offset)
		}
		successWriter.Close()
	}()
}
1077332
func (b *Buffer) ReadAndSubscribe(msgc chan<- *rfc5424.Message, donec <-chan struct{}) []*rfc5424.Message {
	b.mu.RLock()
	defer b.mu.RUnlock()

	b.subscribe(msgc, donec)
	return b.read()
}
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()
		}
	}
}
884640
func readReply(conn *Conn) (*Reply, error) {
	line, err := readLine(conn)
	if err != nil {
		return nil, err
	}
	if len(line) == 0 {
		return nil, ErrRead
	}
	switch line[0] {
	case '+':
		switch {
		case len(line) == 3 && line[1] == 'O' && line[2] == 'K':
			return okReply, nil
		case len(line) == 5 && line[1] == 'P' && line[2] == 'O' && line[3] == 'N' && line[4] == 'G':
			return pongReply, nil
		case len(line) == 7 && line[1] == 'Q' && line[2] == 'U' && line[3] == 'E' &&
			line[4] == 'U' && line[5] == 'E' && line[6] == 'D':
			return queuedReply, nil
		default:
			return &Reply{
				replyType:   ReplyStatus,
				stringValue: line[1:],
			}, nil

		}
	case '-':
		return &Reply{
			replyType:   ReplyError,
			stringValue: line[1:],
		}, nil
	case ':':
		intValue, err := strconv.ParseInt(string(line[1:]), 10, 64)
		if err != nil {
			return nil, ErrRead
		}
		return &Reply{
			replyType:    ReplyInteger,
			integerValue: intValue,
		}, nil
	case '$':
		l, err := strconv.ParseInt(string(line[1:]), 10, 64)
		if err != nil {
			return nil, ErrRead
		}
		if l < 0 {
			return &Reply{
				replyType: ReplyNil,
			}, nil
		}
		b := make([]byte, l)
		_, err = io.ReadFull(conn.rb, b)
		if err != nil {
			return nil, ErrRead
		}
		line, err = readLine(conn)
		if err != nil || len(line) != 0 {
			return nil, ErrRead
		}
		return &Reply{
			replyType:   ReplyString,
			stringValue: b,
		}, nil
	case '*':
		l, err := strconv.ParseInt(string(line[1:]), 10, 64)
		if err != nil {
			return nil, ErrRead
		}
		if l < 0 {
			return &Reply{
				replyType: ReplyNil,
			}, nil
		}
		replyArray := make([]*Reply, l)
		for i := range replyArray {
			replyArray[i], err = readReply(conn)
			if err != nil {
				return nil, err
			}
		}
		return &Reply{
			replyType:  ReplyArray,
			arrayValue: replyArray,
		}, nil
	default:
		return nil, ErrRead
	}
}
941273
func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, subject, reply []byte, flags int) [][]byte {
	var queues [][]byte
	// msg header for clients.
	msgh := c.msgb[1:msgHeadProtoLen]
	msgh = append(msgh, subject...)
	msgh = append(msgh, ' ')
	si := len(msgh)

	// For sending messages across routes. Reset it if we have one.
	// We reuse this data structure.
	if c.in.rts != nil {
		c.in.rts = c.in.rts[:0]
	}

	// Loop over all normal subscriptions that match.
	for _, sub := range r.psubs {
		// Check if this is a send to a ROUTER. We now process
		// these after everything else.
		switch sub.client.kind {
		case ROUTER:
			if c.kind == ROUTER {
				continue
			}
			c.addSubToRouteTargets(sub)
			continue
		case GATEWAY:
			// Never send to gateway from here.
			continue
		case LEAF:
			// We handle similarly to routes and use the same data structures.
			// Leaf node delivery audience is different however.
			// Also leaf nodes are always no echo, so we make sure we are not
			// going to send back to ourselves here.
			if c != sub.client {
				c.addSubToRouteTargets(sub)
			}
			continue
		}
		// Check for stream import mapped subs. These apply to local subs only.
		if sub.im != nil && sub.im.prefix != "" {
			// Redo the subject here on the fly.
			msgh = c.msgb[1:msgHeadProtoLen]
			msgh = append(msgh, sub.im.prefix...)
			msgh = append(msgh, subject...)
			msgh = append(msgh, ' ')
			si = len(msgh)
		}
		// Normal delivery
		mh := c.msgHeader(msgh[:si], sub, reply)
		c.deliverMsg(sub, mh, msg)
	}

	// Set these up to optionally filter based on the queue lists.
	// This is for messages received from routes which will have directed
	// guidance on which queue groups we should deliver to.
	qf := c.pa.queues

	// For all non-client connections, we may still want to send messages to
	// leaf nodes or routes even if there are no queue filters since we collect
	// them above and do not process inline like normal clients.
	if c.kind != CLIENT && qf == nil {
		// However, if this is a gateway connection which should be treated
		// as a client, still go and pick queue subscriptions, otherwise
		// jump to sendToRoutesOrLeafs.
		if !(c.kind == GATEWAY && (flags&pmrTreatGatewayAsClient != 0)) {
			goto sendToRoutesOrLeafs
		}
	}

	// Check to see if we have our own rand yet. Global rand
	// has contention with lots of clients, etc.
	if c.in.prand == nil {
		c.in.prand = rand.New(rand.NewSource(time.Now().UnixNano()))
	}

	// Process queue subs
	for i := 0; i < len(r.qsubs); i++ {
		qsubs := r.qsubs[i]
		// If we have a filter check that here. We could make this a map or someting more
		// complex but linear search since we expect queues to be small. Should be faster
		// and more cache friendly.
		if qf != nil && len(qsubs) > 0 {
			tqn := qsubs[0].queue
			for _, qn := range qf {
				if bytes.Equal(qn, tqn) {
					goto selectQSub
				}
			}
			continue
		}

	selectQSub:

		// We will hold onto remote or lead qsubs when we are coming from
		// a route or a leaf node just in case we can no longer do local delivery.
		var rsub *subscription

		// Find a subscription that is able to deliver this message
		// starting at a random index.
		startIndex := c.in.prand.Intn(len(qsubs))
		for i := 0; i < len(qsubs); i++ {
			index := (startIndex + i) % len(qsubs)
			sub := qsubs[index]
			if sub == nil {
				continue
			}
			kind := sub.client.kind
			// Potentially sending to a remote sub across a route or leaf node.
			if kind == ROUTER || kind == LEAF {
				if c.kind == ROUTER || c.kind == LEAF || (c.kind == CLIENT && kind == LEAF) {
					// We just came from a route/leaf, so skip and prefer local subs.
					// Keep our first rsub in case all else fails.
					if rsub == nil {
						rsub = sub
					}
					continue
				} else {
					c.addSubToRouteTargets(sub)
					if flags&pmrCollectQueueNames != 0 {
						queues = append(queues, sub.queue)
					}
				}
				break
			}

			// Check for mapped subs
			if sub.im != nil && sub.im.prefix != "" {
				// Redo the subject here on the fly.
				msgh = c.msgb[1:msgHeadProtoLen]
				msgh = append(msgh, sub.im.prefix...)
				msgh = append(msgh, subject...)
				msgh = append(msgh, ' ')
				si = len(msgh)
			}

			mh := c.msgHeader(msgh[:si], sub, reply)
			if c.deliverMsg(sub, mh, msg) {
				// Clear rsub
				rsub = nil
				if flags&pmrCollectQueueNames != 0 {
					queues = append(queues, sub.queue)
				}
				break
			}
		}

		if rsub != nil {
			// If we are here we tried to deliver to a local qsub
			// but failed. So we will send it to a remote or leaf node.
			c.addSubToRouteTargets(rsub)
			if flags&pmrCollectQueueNames != 0 {
				queues = append(queues, rsub.queue)
			}
		}
	}

sendToRoutesOrLeafs:

	// If no messages for routes return here.
	if len(c.in.rts) == 0 {
		return queues
	}

	// We address by index to avoid struct copy.
	// We have inline structs for memory layout and cache coherency.
	for i := range c.in.rts {
		rt := &c.in.rts[i]
		kind := rt.sub.client.kind
		mh := c.msgb[:msgHeadProtoLen]
		if kind == ROUTER {
			// Router (and Gateway) nodes are RMSG. Set here since leafnodes may rewrite.
			mh[0] = 'R'
			mh = append(mh, acc.Name...)
			mh = append(mh, ' ')
		} else {
			// Leaf nodes are LMSG
			mh[0] = 'L'
			// Remap subject if its a shadow subscription, treat like a normal client.
			if rt.sub.im != nil && rt.sub.im.prefix != "" {
				mh = append(mh, rt.sub.im.prefix...)
			}
		}
		mh = append(mh, subject...)
		mh = append(mh, ' ')

		if len(rt.qs) > 0 {
			if reply != nil {
				mh = append(mh, "+ "...) // Signal that there is a reply.
				mh = append(mh, reply...)
				mh = append(mh, ' ')
			} else {
				mh = append(mh, "| "...) // Only queues
			}
			mh = append(mh, rt.qs...)
		} else if reply != nil {
			mh = append(mh, reply...)
			mh = append(mh, ' ')
		}
		mh = append(mh, c.pa.szb...)
		mh = append(mh, _CRLF_...)
		c.deliverMsg(rt.sub, mh, msg)
	}
	return queues
}
49088
func (conn *BytesManualConnectionStr) ConsumePartition(msgClb func(message *client.ConsumerMessage), topic string, partition int32, offset int64) error {
	conn.multiplexer.rwlock.Lock()
	defer conn.multiplexer.rwlock.Unlock()
	var err error

	// check if we have already consumed the topic on partition and offset
	var found bool
	var subs *consumerSubscription

	for _, subscription := range conn.multiplexer.mapping {
		if subscription.manual == false {
			// do not mix dynamic and manual mode
			continue
		}
		if subscription.topic == topic && subscription.partition == partition && subscription.offset == offset {
			found = true
			subs = subscription
			break
		}
	}

	if !found {
		subs = &consumerSubscription{
			manual:         true, // manual example
			topic:          topic,
			partition:      partition,
			offset:         offset,
			connectionName: conn.name,
			byteConsMsg:    msgClb,
		}
		// subscribe new topic on partition
		conn.multiplexer.mapping = append(conn.multiplexer.mapping, subs)
	}

	// add subscription to consumerList
	subs.byteConsMsg = msgClb

	if conn.multiplexer.started {
		conn.multiplexer.Infof("Starting 'post-init' manual Consumer")
		subs.partitionConsumer, err = conn.StartPostInitConsumer(topic, partition, offset)
		if err != nil {
			return err
		}
		if subs.partitionConsumer == nil {
			return nil
		}
	}

	return nil
}
961284
func (mr *MockConnectorMockRecorder) QueueSubscribe(ctx, name, subject, group, output interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "QueueSubscribe", reflect.TypeOf((*MockConnector)(nil).QueueSubscribe), ctx, name, subject, group, output)
}
1373924
func (c *Client) Subscribe(topic string, qos byte, callback MessageHandler) Token {
	token := newToken(packets.Subscribe).(*SubscribeToken)
	DEBUG.Println(CLI, "enter Subscribe")
	if !c.IsConnected() {
		token.err = ErrNotConnected
		token.flowComplete()
		return token
	}
	sub := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
	if err := validateTopicAndQos(topic, qos); err != nil {
		token.err = err
		return token
	}
	sub.Topics = append(sub.Topics, topic)
	sub.Qoss = append(sub.Qoss, qos)
	DEBUG.Println(sub.String())

	if callback != nil {
		c.msgRouter.addRoute(topic, callback)
	}

	token.subs = append(token.subs, topic)
	c.oboundP <- &PacketAndToken{p: sub, t: token}
	DEBUG.Println(CLI, "exit Subscribe")
	return token
}