// Subscribe to notifications.
// `unsubscribe` releases resources associated with the subscription and should be called asap.
485266
func (d *IdentifyDispatch) Subscribe(mctx MetaContext) (unsubscribe func(), recvCh <-chan IdentifyDispatchMsg) {
	mctx.Debug("IdentifyDispatch.Subscribe")
	ch := make(chan IdentifyDispatchMsg, 10)
	d.Lock()
	defer d.Unlock()
	d.listeners = append(d.listeners, ch)
	unsubscribe = func() {
		mctx.Debug("IdentifyDispatch.Unsubcribe")
		d.Lock()
		defer d.Unlock()
		var listeners []chan<- IdentifyDispatchMsg
		for _, ch2 := range d.listeners {
			if ch == ch2 {
				continue
			}
			listeners = append(listeners, ch2)
		}
		d.listeners = listeners
	}
	return unsubscribe, ch
}
1301775
func (mr *MockConnectorMockRecorder) Unsubscribe(name interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Unsubscribe", reflect.TypeOf((*MockConnector)(nil).Unsubscribe), name)
}
516118
func (s *Subscriber) Subscribe(c string) int {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.channels[c] = struct{}{}
	return s.count()
}
612662
func (n *Notifier) takeSubscription() *Subscription {
	n.mu.Lock()
	defer n.mu.Unlock()
	n.callReturned = true
	return n.sub
}
1260128
func (sub *Subscription) Unsubscribe() {
	sub.unsubOnce.Do(func() {
	uninstallLoop:
		for {
			// write uninstall request and consume logs/hashes. This prevents
			// the eventLoop broadcast method to deadlock when writing to the
			// filter event channel while the subscription loop is waiting for
			// this method to return (and thus not reading these events).
			select {
			case sub.es.uninstall <- sub.f:
				break uninstallLoop
			case <-sub.f.logs:
			case <-sub.f.hashes:
			case <-sub.f.headers:
			}
		}

		// wait for filter to be uninstalled in work loop before returning
		// this ensures that the manager won't use the event channel which
		// will probably be closed by the client asap after this method returns.
		<-sub.Err()
	})
}
629796
func (c *Controller) Unsubscribe(name string) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	sub, ok := c.subscriptions[name]
	if !ok {
		return fmt.Errorf("Unknown subscription '%s'", name)
	}
	msg := NewMsg(MsgCodeStop, name, sub.address)
	smsg, err := rlp.EncodeToBytes(msg)
	if err != nil {
		return err
	}
	err = c.pss.SendAsym(sub.pubkeyId, controlTopic, smsg)
	if err != nil {
		return err
	}
	delete(c.subscriptions, name)
	return 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()
		}
	}
}
1212183
func (logger *StreamLogger) Unsubscribe(ch chan interface{}) {
	logger.mu.Lock()
	defer logger.mu.Unlock()

	delete(logger.subscribed, ch)
}
85111
func NewUNSUBSCRIBE(opts *UNSUBSCRIBEOptions) (Packet, error) {
	// Initialize the options.
	if opts == nil {
		opts = &UNSUBSCRIBEOptions{}
	}

	// Validate the options.
	if err := opts.validate(); err != nil {
		return nil, err
	}

	// Create a SUBSCRIBE Packet.
	p := &UNSUBSCRIBE{
		PacketID:     opts.PacketID,
		TopicFilters: opts.TopicFilters,
	}

	// Set the variable header to the Packet.
	p.setVariableHeader()

	// Set the payload to the Packet.
	p.setPayload()

	// Set the Fixed header to the Packet.
	p.setFixedHeader()

	// Return the Packet.
	return p, nil
}
990764
func (em *Emitter) Subscribe(ctx context.Context, subscriber string, queryable query.Queryable, bufferSize int) (<-chan interface{}, error) {
	qry, err := queryable.Query()
	if err != nil {
		return nil, err
	}
	return em.pubsubServer.Subscribe(ctx, subscriber, qry, bufferSize)
}
202670
func (w *wrapper) Unsubscribe(c Committer) {
	w.mut.Lock()
	for i := range w.subs {
		if w.subs[i] == c {
			copy(w.subs[i:], w.subs[i+1:])
			w.subs[len(w.subs)-1] = nil
			w.subs = w.subs[:len(w.subs)-1]
			break
		}
	}
	w.mut.Unlock()
}
721955
func unsubscribe(pid uint32) {
	lumber.Trace("Removing proxy from subscribers...")

	mutex.Lock()
	delete(subscribers, pid)
	mutex.Unlock()
}
236720
func (w *wrapper) Subscribe(c Committer) {
	w.mut.Lock()
	w.subs = append(w.subs, c)
	w.mut.Unlock()
}
896491
func (em *Emitter) Unsubscribe(ctx context.Context, subscriber string, queryable query.Queryable) error {
	pubsubQuery, err := queryable.Query()
	if err != nil {
		return nil
	}
	return em.pubsubServer.Unsubscribe(ctx, subscriber, pubsubQuery)
}
1267009
func (n *Nats) UnsubscribeAll() {
	for _, route := range n.Opts.routes {
		route.subsc.Unsubscribe()
	}
}
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
}
1233451
func (c *Client) Subscribe(listener *Listener) {
	listener.Active = true
	c.tasks <- task{Listener: listener, Action: subscribeAction}
}