// ReallySubscribeExchanges stores the client and their exchange subscriptions.
// It waits for the client contexts Done channel to close and deletes the client
// from the RateServer.clients map.
353989
func (server *RateServer) ReallySubscribeExchanges(hello *dcrrates.ExchangeSubscription, stream GRPCStream) (err error) {
	// For now, require the ExchangeBot clients to have the same base currency.
	// ToDo: Allow any index.
	if hello.BtcIndex != server.btcIndex {
		return fmt.Errorf("Exchange subscription has wrong BTC index. Given: %s, Required: %s", hello.BtcIndex, server.btcIndex)
	}
	// Save the client for use in the main loop.
	client, sid := server.addClient(stream, hello)

	// Get the address for an Infof. Seems like peerInfo is always found, but
	// am checking anyway.
	peerInfo, peerFound := grpcPeer.FromContext(client.Stream().Context())
	var clientAddr string
	if peerFound {
		clientAddr = peerInfo.Addr.String()
		log.Infof("Client has connected from %s", clientAddr)
	}

	state := server.xcBot.State()
	// Send Decred exchanges.
	err = sendStateList(client, state.DcrBtc)
	if err != nil {
		return err
	}
	// Send Bitcoin-fiat indices.
	for token := range state.FiatIndices {
		err = client.SendExchangeUpdate(&dcrrates.ExchangeRateUpdate{
			Token:   token,
			Indices: server.xcBot.Indices(token),
		})
		if err != nil {
			log.Errorf("Error encountered while sending fiat indices to client at %s: %v", clientAddr, err)
			// Assuming the Done channel will be closed on error, no further iteration
			// is necessary.
			break
		}
	}

	// Keep stream alive.
	<-client.Stream().Context().Done()

	server.deleteClient(sid)
	log.Infof("Client at %s has disconnected", clientAddr)
	return
}
219690
func (nc *Conn) unsubscribe(sub *Subscription, max int, drainMode bool) error {
	nc.mu.Lock()
	// ok here, but defer is expensive
	defer nc.mu.Unlock()
	defer nc.kickFlusher()

	if nc.isClosed() {
		return ErrConnectionClosed
	}

	nc.subsMu.RLock()
	s := nc.subs[sub.sid]
	nc.subsMu.RUnlock()
	// Already unsubscribed
	if s == nil {
		return nil
	}

	maxStr := _EMPTY_
	if max > 0 {
		s.max = uint64(max)
		maxStr = strconv.Itoa(max)
	} else if !drainMode {
		nc.removeSub(s)
	}

	if drainMode {
		go nc.checkDrained(sub)
	}

	// We will send these for all subs when we reconnect
	// so that we can suppress here.
	if !nc.isReconnecting() {
		fmt.Fprintf(nc.bw, unsubProto, s.sid, maxStr)
	}
	return nil
}
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
}
371820
func (self *Client) Subscribe(channel string) (ch *Channel) {
	for _, ch := range self.Channels {
		if ch.Name == channel {
			self._subscribe <- ch
			return ch
		}
	}
	ch = &Channel{Name: channel, bindings: &self.bindings}
	self._subscribe <- ch
	return
}
664593
func NeighSubscribe(ch chan<- NeighUpdate, done <-chan struct{}) error {
	return neighSubscribeAt(netns.None(), netns.None(), ch, done, nil, false)
}
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
}
439884
func (m *MemoryBackend) Unsubscribe(client *Client, topics []string, ack Ack) error {
	// delete subscriptions
	for _, t := range topics {
		client.Session().(*memorySession).subscriptions.Empty(t)
	}

	// call ack if provided
	if ack != nil {
		ack()
	}

	return nil
}
1234006
func (s *Server) subscriptionHandler() {
	defer s.wg.Done()

	for {
		select {

		// If a client update is received, the either a new
		// subscription becomes active, or we cancel and existing one.
		case update := <-s.clientUpdates:
			clientID := update.clientID

			// In case this is a cancellation, stop the client's
			// underlying queue, and remove the client from the set
			// of active subscription clients.
			if update.cancel {
				client, ok := s.clients[update.clientID]
				if ok {
					client.updates.Stop()
					close(client.quit)
					delete(s.clients, clientID)
				}

				continue
			}

			// If this was not a cancellation, start the underlying
			// queue and add the client to our set of subscription
			// clients. It will be notified about any new updates
			// the server receives.
			update.client.updates.Start()
			s.clients[update.clientID] = update.client

		// A new update was received, forward it to all active clients.
		case upd := <-s.updates:
			for _, client := range s.clients {
				select {
				case client.updates.ChanIn() <- upd:
				case <-client.quit:
				case <-s.quit:
					return
				}
			}

		// In case the server is shutting down, stop the clients and
		// close the quit channels to notify them.
		case <-s.quit:
			for _, client := range s.clients {
				client.updates.Stop()
				close(client.quit)
			}
			return
		}
	}
}
694455
func (s *SubSocket) Unsubscribe(topic string) error {
	return s.sock.SetOption(mangos.OptionUnsubscribe, topic)
}
259307
func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, nil, ch, false)
}
1182612
func (s *Service) Unsubscribe(topic string) GenericFuture {
	return s.UnsubscribeMultiple([]string{topic})
}
1255856
func (e *EventListener) Unsubscribe(s *EventSubscriber) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if subs, ok := e.subscribers[s.appID]; ok {
		delete(subs, s)
		if len(subs) == 0 {
			delete(e.subscribers, s.appID)
		}
	}
}
1314622
func (tsm *TargetStatsMultiplexer) Unsubscribe(i int) error {
	c, ok := tsm.listeners[i]
	if !ok {
		return fmt.Errorf("TargetStatsMultiplexer.Unsubscribe(%v): not suc channel", i)
	}
	delete(tsm.listeners, i)
	close(c)
	return nil
}
582186
func (exchange *Exchange) Unsubscribe(handle Handle) error {
	exchange.cond.L.Lock()
	defer exchange.cond.L.Unlock()

	if exchange.state != stateRunning {
		return ErrInvalidState
	}

	topic, ok := exchange.topicForHandle[handle]
	if !ok {
		return ErrInvalidHandle
	}

	delete(exchange.topicForHandle, handle)

	item := exchange.trie.Get(patricia.Prefix(topic))
	if item == nil {
		return ErrInvalidHandle
	}

	records := item.(*[]*handlerRecord)
	if len(*records) == 1 {
		if exchange.trie.Delete(patricia.Prefix(topic)) {
			return nil
		}
	} else {
		for i, record := range *records {
			if record.handle == handle {
				*records = append((*records)[:i], (*records)[i+1:]...)
				return nil
			}
		}
	}

	return ErrInvalidHandle
}
489574
func (c *PubSub) Unsubscribe(channels ...string) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	for _, channel := range channels {
		delete(c.channels, channel)
	}
	err := c.subscribe("unsubscribe", channels...)
	return err
}
567681
func (p *pluginControl) UnsubscribeDeps(id string) []serror.SnapError {
	// update view and unsubscribe to plugins
	return p.subscriptionGroups.Remove(id)
}
349524
func (s *Subscriber) Unsubscribe(c string) int {
	s.mu.Lock()
	defer s.mu.Unlock()

	delete(s.channels, c)
	return s.count()
}