// enter 'subscribed state', or return the existing one.
327035
func (m *Miniredis) subscribedState(c *server.Peer) *Subscriber {
	ctx := getCtx(c)
	sub := ctx.subscriber
	if sub != nil {
		return sub
	}

	sub = newSubscriber()
	m.addSubscriber(sub)

	c.OnDisconnect(func() {
		m.Lock()
		m.removeSubscriber(sub)
		m.Unlock()
	})

	ctx.subscriber = sub

	go monitorPublish(c, sub.publish)

	return sub
}
1226714
func (s *SubServer) Subscribe(serviceURL, sid string, timeoutInSeconds int, autoRefresh bool, sub Subscriber) (string, error) {
	req, err := http.NewRequest(
		"SUBSCRIBE",
		serviceURL,
		nil)

	if err != nil {
		return "", err
	}

	req.Header.Add("TIMEOUT", "Second-"+strconv.Itoa(timeoutInSeconds))
	req.Close = true

	if sid != "" {
		// Renew subscription
		req.Header.Add("SID", sid)
	} else {
		req.Header.Add("CALLBACK", "<http://"+s.listenAddr+">")
		req.Header.Add("NT", "upnp:event")
	}

	client := &http.Client{}
	resp, err := client.Do(req)
	if resp != nil {
		defer resp.Body.Close()
	}
	if err != nil {
		return "", err
	}

	sids, ok := resp.Header["Sid"]
	if !ok || len(sids) == 0 {
		return "", errors.New("Sid header missing in response")
	}

	// If the sid == "" this is a new subscribe otherwise it is a refresh in
	// which case we don't need to insert this book keeping
	if sid == "" {
		s.subs[sids[0]] = &subInfo{
			SID:              sids[0],
			ServiceURL:       serviceURL,
			Subscriber:       sub,
			TimeoutInSeconds: timeoutInSeconds,
		}
	}

	if autoRefresh {
		s.renewRefreshTimer(sids[0])
	}

	return sids[0], nil
}
1013847
func (c *Local) resubscribe(subscriber string, q tmpubsub.Query) types.Subscription {
	attempts := 0
	for {
		if !c.IsRunning() {
			return nil
		}

		sub, err := c.EventBus.Subscribe(context.Background(), subscriber, q)
		if err == nil {
			return sub
		}

		attempts++
		time.Sleep((10 << uint(attempts)) * time.Millisecond) // 10ms -> 20ms -> 40ms
	}
}
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
			}
		}
	}
}
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
}
481377
func (c *Conn) SubscribeUnitsCustom(interval time.Duration, buffer int, isChanged func(*UnitStatus, *UnitStatus) bool, filterUnit func(string) bool) (<-chan map[string]*UnitStatus, <-chan error) {
	old := make(map[string]*UnitStatus)
	statusChan := make(chan map[string]*UnitStatus, buffer)
	errChan := make(chan error, buffer)

	go func() {
		for {
			timerChan := time.After(interval)

			units, err := c.ListUnits()
			if err == nil {
				cur := make(map[string]*UnitStatus)
				for i := range units {
					if filterUnit != nil && filterUnit(units[i].Name) {
						continue
					}
					cur[units[i].Name] = &units[i]
				}

				// add all new or changed units
				changed := make(map[string]*UnitStatus)
				for n, u := range cur {
					if oldU, ok := old[n]; !ok || isChanged(oldU, u) {
						changed[n] = u
					}
					delete(old, n)
				}

				// add all deleted units
				for oldN := range old {
					changed[oldN] = nil
				}

				old = cur

				if len(changed) != 0 {
					statusChan <- changed
				}
			} else {
				errChan <- err
			}

			<-timerChan
		}
	}()

	return statusChan, errChan
}
13472
func (e *Events) Subscribe() ([]eventtypes.Message, chan interface{}, func()) {
	eventSubscribers.Inc()
	e.mu.Lock()
	current := make([]eventtypes.Message, len(e.events))
	copy(current, e.events)
	l := e.pub.Subscribe()
	e.mu.Unlock()

	cancel := func() {
		e.Evict(l)
	}
	return current, l, cancel
}
789019
func (s *NatsSubscriber) Subscribe(subject, group string, handler nats.Handler) error {
	ch, err := s.server.initSubject(subject)
	if err != nil {
		return err
	}

	go func(s *NatsSubscriber, subject string, handler nats.Handler) {
		for r := range ch {
			handler.Process(&natsSubscription{subject: subject},
				&natsMessage{
					r: r,
				},
			)
		}
	}(s, subject, handler)
	return nil
}
446419
func (c *Client) Subscribe(event string) (*pstypes.WebSocketMessage, error) {
	// Validate the event type.
	_, _, ok := pstypes.ValidateSubscription(event)
	if !ok {
		return nil, fmt.Errorf("invalid subscription %s", event)
	}

	// Send the subscribe message.
	msg := newSubscribeMsg(event)
	_ = c.SetWriteDeadline(time.Now().Add(c.WriteTimeout))
	_, err := c.Write([]byte(msg))
	if err != nil {
		return nil, fmt.Errorf("failed to send subscribe message: %v", err)
	}

	// Read the response.
	return c.ReceiveMsg()
}
611085
func (m *Monitor) Subscribe() chan runc.Exit {
	c := make(chan runc.Exit, bufferSize)
	m.Lock()
	m.subscribers[c] = struct{}{}
	m.Unlock()
	return c
}
542885
func (c *Client) Subscribe(topic string, fn EventHandler) error {
	id := NewID()
	c.registerListener(id)
	sub := &Subscribe{
		Request: id,
		Options: make(map[string]interface{}),
		Topic:   URI(topic),
	}
	if err := c.Send(sub); err != nil {
		return err
	}
	// wait to receive SUBSCRIBED message
	msg, err := c.waitOnListener(id)
	if err != nil {
		return err
	} else if e, ok := msg.(*Error); ok {
		return fmt.Errorf("error subscribing to topic '%v': %v", topic, e.Error)
	} else if subscribed, ok := msg.(*Subscribed); !ok {
		return fmt.Errorf(formatUnexpectedMessage(msg, SUBSCRIBED))
	} else {
		// register the event handler with this subscription
		c.events[subscribed.Subscription] = &eventDesc{topic, fn}
	}
	return nil
}
1019168
func (rcvr *Receiver) Subscribe(svcName string) {
	for _, subName := range rcvr.Subscriptions {
		if subName == svcName {
			return
		}
	}

	rcvr.Subscriptions = append(rcvr.Subscriptions, svcName)
}
657656
func (m *Miniredis) NewSubscriber() *Subscriber {
	sub := newSubscriber()

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

	return sub
}
627659
func (tm *topicManager) IsSubscribed(topic string) bool {
	tm.locker.RLock()
	defer tm.locker.RUnlock()
	return tm.allTopics[topic] != nil
}
266196
func (s *SubscriptionSet) Subscribe() (<-chan map[string]*UnitStatus, <-chan error) {
	// TODO: Make fully evented by using systemd 209 with properties changed values
	return s.conn.SubscribeUnitsCustom(time.Second, 0,
		mismatchUnitStatus,
		func(unit string) bool { return s.filter(unit) },
	)
}
520541
func (p *Peer) Subscribe(sub peer.Subscriber) {
	p.lock.Lock()
	p.subscribers[sub] = struct{}{}
	p.lock.Unlock()
}
544915
func (w *WebSocketService) Subscribe() error {
	// Subscribe to each channel
	if err := w.sendSubscribeMessages(); err != nil {
		return err
	}

	for {
		_, p, err := w.ws.ReadMessage()
		if err != nil {
			return err
		}

		if bytes.Contains(p, []byte("event")) {
			w.handleEventMessage(p)
		} else {
			w.handleDataMessage(p)
		}
	}
	// nolint
	return nil
}