// RouteSubscribe takes a chan down which notifications will be sent
// when routes are added or deleted. Close the 'done' chan to stop subscription.
333215
func RouteSubscribe(ch chan<- RouteUpdate, done <-chan struct{}) error {
	return routeSubscribeAt(netns.None(), netns.None(), ch, done, nil, false)
}
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
}
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
}
394916
func (p *Proxy) Subscribe(tags []string) {
	lumber.Trace("Proxy subscribing to '%s'...", tags)

	if len(tags) == 0 {
		return
	}

	// add proxy to subscribers list here so not all clients are 'subscribers'
	// since gets added to a map, there are no duplicates
	subscribe(p)

	// add tags to subscription
	p.Lock()
	p.subscriptions.Add(tags)
	p.Unlock()
}
902059
func (c *client) unsubscribe(acc *Account, sub *subscription, force bool) {
	c.mu.Lock()
	if !force && sub.max > 0 && sub.nm < sub.max {
		c.Debugf(
			"Deferring actual UNSUB(%s): %d max, %d received",
			string(sub.subject), sub.max, sub.nm)
		c.mu.Unlock()
		return
	}
	c.traceOp("<-> %s", "DELSUB", sub.sid)

	delete(c.subs, string(sub.sid))
	if c.kind != CLIENT && c.kind != SYSTEM {
		c.removeReplySubTimeout(sub)
	}

	if acc != nil {
		acc.sl.Remove(sub)
	}

	// Check to see if we have shadow subscriptions.
	var updateRoute bool
	shadowSubs := sub.shadow
	sub.shadow = nil
	if len(shadowSubs) > 0 {
		updateRoute = (c.kind == CLIENT || c.kind == SYSTEM || c.kind == LEAF) && c.srv != nil
	}
	c.mu.Unlock()

	for _, nsub := range shadowSubs {
		if err := nsub.im.acc.sl.Remove(nsub); err != nil {
			c.Debugf("Could not remove shadow import subscription for account %q", nsub.im.acc.Name)
		} else if updateRoute {
			c.srv.updateRouteSubscriptionMap(nsub.im.acc, nsub, -1)
		}
		// Now check on leafnode updates.
		c.srv.updateLeafNodes(nsub.im.acc, nsub, -1)
	}
}
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) },
	)
}
476822
func (mgr *SettingsManager) Subscribe(subCh chan<- *ArgoCDSettings) {
	mgr.mutex.Lock()
	defer mgr.mutex.Unlock()
	mgr.subscribers = append(mgr.subscribers, subCh)
	log.Infof("%v subscribed to settings updates", subCh)
}
948414
func AddrSubscribe(ch chan<- AddrUpdate, done <-chan struct{}) error {
	return addrSubscribeAt(netns.None(), netns.None(), ch, done, nil, false)
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
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
}
545987
func (s *Server) sendQueueSubOrUnsubToGateways(accName string, qsub *subscription, added bool) {
	if qsub.queue == nil {
		return
	}

	gwsa := [16]*client{}
	gws := gwsa[:0]
	s.getInboundGatewayConnections(&gws)
	if len(gws) == 0 {
		return
	}

	var protoa [512]byte
	var proto []byte
	for _, c := range gws {
		if proto == nil {
			proto = protoa[:0]
			if added {
				proto = append(proto, rSubBytes...)
			} else {
				proto = append(proto, rUnsubBytes...)
			}
			proto = append(proto, accName...)
			proto = append(proto, ' ')
			proto = append(proto, qsub.subject...)
			proto = append(proto, ' ')
			proto = append(proto, qsub.queue...)
			if added {
				// For now, just use 1 for the weight
				proto = append(proto, ' ', '1')
			}
			proto = append(proto, CR_LF...)
		}
		c.mu.Lock()
		c.sendProto(proto, false)
		if c.trace {
			c.traceOutOp("", proto[:len(proto)-LEN_CR_LF])
		}
		c.mu.Unlock()
	}
}
611085
func (m *Monitor) Subscribe() chan runc.Exit {
	c := make(chan runc.Exit, bufferSize)
	m.Lock()
	m.subscribers[c] = struct{}{}
	m.Unlock()
	return c
}
919496
func (r *marathonClient) Subscribe(callback string) error {
	path := fmt.Sprintf("%s?callbackUrl=%s", marathonAPISubscription, callback)
	return r.apiPost(path, "", nil)

}
1264673
func handleSubscribe(proxy *mist.Proxy, msg mist.Message) error {
	proxy.Subscribe(msg.Tags)
	return nil
}
1330581
func (e *eventer) Subscribe() eventChannel {
	e.eventsMutex.Lock()
	defer e.eventsMutex.Unlock()
	out := make(eventChannel, eventChanBufferSize)
	e.outs[out] = out
	return out
}
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()
}
780948
func subscribe(p *Proxy) {
	lumber.Trace("Adding proxy to subscribers...")

	mutex.Lock()
	subscribers[p.id] = p
	mutex.Unlock()
}