// SubscriberAfter functions are executed on the subscriber reply after the
// endpoint is invoked, but before anything is published to the reply.
509586
func SubscriberAfter(after ...SubscriberResponseFunc) SubscriberOption {
	return func(s *Subscriber) { s.after = append(s.after, after...) }
}
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
}
724536
func (s *Subscriber) Subscribe(e Event, f interface{}) *Subscriber {
	unsubscribe := s.events.subscribe(e, f)
	s.unsuscribes = append(s.unsuscribes, unsubscribe)
	return s
}
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
}
408345
func (bc *brokerConsumer) subscriptionConsumer() {
	<-bc.wait // wait for our first piece of work

	for newSubscriptions := range bc.newSubscriptions {
		bc.updateSubscriptions(newSubscriptions)

		if len(bc.subscriptions) == 0 {
			// We're about to be shut down or we're about to receive more subscriptions.
			// Either way, the signal just hasn't propagated to our goroutine yet.
			<-bc.wait
			continue
		}

		response, err := bc.fetchNewMessages()

		if err != nil {
			Logger.Printf("consumer/broker/%d disconnecting due to error processing FetchRequest: %s\n", bc.broker.ID(), err)
			bc.abort(err)
			return
		}

		bc.acks.Add(len(bc.subscriptions))
		for child := range bc.subscriptions {
			child.feeder <- response
		}
		bc.acks.Wait()
		bc.handleResponses()
	}
}
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
}
627659
func (tm *topicManager) IsSubscribed(topic string) bool {
	tm.locker.RLock()
	defer tm.locker.RUnlock()
	return tm.allTopics[topic] != nil
}
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
}
592549
func (p *PubSub) Subscribe(id string, s Subscriber) error {
	p.m.Lock()
	defer p.m.Unlock()

	p.subscribers[id] = s
	return nil
}
356561
func NewSubscriberWrapper() server.SubscriberWrapper {
	return func(fn server.SubscriberFunc) server.SubscriberFunc {
		return func(ctx context.Context, p server.Message) (err error) {
			t := newPublicationTracker(p, ServerProfile)
			ctx = t.start(ctx, false)

			defer func() { t.end(ctx, err) }()

			spanCtx := getTraceFromCtx(ctx)
			if spanCtx != nil {
				ctx, t.span = trace.StartSpanWithRemoteParent(
					ctx,
					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
					*spanCtx,
				)
			} else {
				ctx, t.span = trace.StartSpan(
					ctx,
					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
				)
			}

			err = fn(ctx, p)
			return
		}
	}
}
718356
func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool) (*Subscription, error) {
	if nc == nil {
		return nil, ErrInvalidConnection
	}
	nc.mu.Lock()
	// ok here, but defer is generally expensive
	defer nc.mu.Unlock()

	// Check for some error conditions.
	if nc.isClosed() {
		return nil, ErrConnectionClosed
	}
	if nc.isDraining() {
		return nil, ErrConnectionDraining
	}

	if cb == nil && ch == nil {
		return nil, ErrBadSubscription
	}

	sub := &Subscription{Subject: subj, Queue: queue, mcb: cb, conn: nc}
	// Set pending limits.
	sub.pMsgsLimit = DefaultSubPendingMsgsLimit
	sub.pBytesLimit = DefaultSubPendingBytesLimit

	// If we have an async callback, start up a sub specific
	// Go routine to deliver the messages.
	if cb != nil {
		sub.typ = AsyncSubscription
		sub.pCond = sync.NewCond(&sub.mu)
		go nc.waitForMsgs(sub)
	} else if !isSync {
		sub.typ = ChanSubscription
		sub.mch = ch
	} else { // Sync Subscription
		sub.typ = SyncSubscription
		sub.mch = ch
	}

	nc.subsMu.Lock()
	nc.ssid++
	sub.sid = nc.ssid
	nc.subs[sub.sid] = sub
	nc.subsMu.Unlock()

	// We will send these for all subs when we reconnect
	// so that we can suppress here if reconnecting.
	if !nc.isReconnecting() {
		fmt.Fprintf(nc.bw, subProto, subj, queue, sub.sid)
		// Kick flusher if needed.
		if len(nc.fch) == 0 {
			nc.kickFlusher()
		}
	}

	return sub, nil
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
1096281
func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Subscription {
	return bc.scope.Track(bc.chainSideFeed.Subscribe(ch))
}
525941
func (fs *FS) SubscribeToObsolete() (<-chan struct{}, error) {
	if err := fs.requireNonEmpty(); err != nil {
		return nil, err
	}

	c := make(chan struct{})
	var once sync.Once
	onHandleChange := folderHandleChangeObserver(
		func() { once.Do(func() { close(c) }) })
	if err := fs.config.Notifier().RegisterForChanges(
		[]data.FolderBranch{fs.root.GetFolderBranch()},
		onHandleChange); err != nil {
		return nil, err
	}
	return c, nil
}
372172
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)

		pubsubHandled.WithLabelValues(
			opts.Topic,
			opts.ServiceName,
			strconv.FormatBool(err == nil),
		).Inc()

		subscriberSize.WithLabelValues(
			opts.Topic,
			opts.ServiceName,
		).Add(float64(len(m.Data)))

		subscribeDurationsHistogram.WithLabelValues(
			opts.Topic,
			opts.ServiceName,
		).Observe(time.Now().Sub(start).Seconds())
		return err
	}
}
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
	}
}
1283943
func WrapSubscriber(w ...server.SubscriberWrapper) Option {
	return func(o *Options) {
		var wrappers []server.Option

		for _, wrap := range w {
			wrappers = append(wrappers, server.WrapSubscriber(wrap))
		}

		// Init once
		o.Server.Init(wrappers...)
	}
}