// SubscribeAt works like Subscribe plus let's the caller choose the network
// namespace in which the socket would be opened (newNs). Then control goes back
// to curNs if open, otherwise to the netns at the time this function was called.
557141
func SubscribeAt(newNs, curNs netns.NsHandle, protocol int, groups ...uint) (*NetlinkSocket, error) {
	c, err := executeInNetns(newNs, curNs)
	if err != nil {
		return nil, err
	}
	defer c()
	return Subscribe(protocol, groups...)
}
434214
func RouteSubscribeAt(ns netns.NsHandle, ch chan<- RouteUpdate, done <-chan struct{}) error {
	return routeSubscribeAt(ns, netns.None(), ch, done, nil, false)
}
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
}
52886
func (c *conn) subscribed(char *ble.Characteristic) {
	h := char.Handle
	if _, found := c.notifiers[h]; found {
		return
	}
	send := func(b []byte) (int, error) {
		c.dev.sendCmd(c.dev.pm, 15, xpc.Dict{
			"kCBMsgArgUUIDs":       [][]byte{},
			"kCBMsgArgAttributeID": h,
			"kCBMsgArgData":        b,
		})
		return len(b), nil
	}
	n := ble.NewNotifier(send)
	c.notifiers[h] = n
	req := ble.NewRequest(c, nil, 0) // convey *conn to user handler.
	go char.NotifyHandler.ServeNotify(req, n)
}
611085
func (m *Monitor) Subscribe() chan runc.Exit {
	c := make(chan runc.Exit, bufferSize)
	m.Lock()
	m.subscribers[c] = struct{}{}
	m.Unlock()
	return c
}
257627
func (c *Client) Subscribe(channel string, m chan<- PubSubMessage, stop <-chan bool) error {
	srv, err := c.selector.PickServer("")
	if err != nil {
		return err
	}
	cn, err := c.getConn(srv)
	if err != nil {
		return err
	}
	// we cannot return this connection to the pool
	// because it'll be in subscribe context.
	//defer cn.condRelease(&err)
	_, err = c.execute(cn.rw, "SUBSCRIBE", channel)
	if err != nil {
		return err
	}
	if err = cn.nc.SetDeadline(time.Time{}); err != nil {
		cn.nc.Close()
		return err
	}
	watcher := make(chan struct{})
	go func() {
		select {
		case <-stop:
			cn.nc.Close()
		case <-watcher:
		}
	}()
	go func() {
		defer cn.nc.Close()
		for {
			raw, err := c.parseResponse(cn.rw.Reader)
			if err != nil {
				m <- PubSubMessage{
					Error: err,
				}
				close(watcher)
				return
			}
			switch raw.(type) {
			case []interface{}:
				ret := raw.([]interface{})
				m <- PubSubMessage{
					Value:   ret[2].(string),
					Channel: ret[1].(string),
					Error:   nil,
				}
			default:
				m <- PubSubMessage{
					Error: ErrServerError,
				}
				close(watcher)
				return
			}
		}
	}()
	return err
}
1330581
func (e *eventer) Subscribe() eventChannel {
	e.eventsMutex.Lock()
	defer e.eventsMutex.Unlock()
	out := make(eventChannel, eventChanBufferSize)
	e.outs[out] = out
	return out
}
476381
func (c *Conn) Subscribe(members ...string) chan *dbus.Signal {
	for _, member := range members {
		c.conn.BusObject().Call("org.freedesktop.DBus.AddMatch", 0,
			fmt.Sprintf("type='signal',interface='org.freedesktop.login1.Manager',member='%s'", member))
	}
	ch := make(chan *dbus.Signal, 10)
	c.conn.Signal(ch)
	return ch
}
1102671
func (s *BrokerNodeInfo) SetClientSubnet(v string) *BrokerNodeInfo {
	s.ClientSubnet = &v
	return s
}
505639
func (s *Server) Subscribe() (*Client, error) {
	// We'll first atomically obtain the next ID for this client from the
	// incrementing client ID counter.
	clientID := atomic.AddUint64(&s.clientCounter, 1)

	// Create the client that will be returned. The Cancel method is
	// populated to send the cancellation intent to the
	// subscriptionHandler.
	client := &Client{
		updates: queue.NewConcurrentQueue(20),
		quit:    make(chan struct{}),
		Cancel: func() {
			select {
			case s.clientUpdates <- &clientUpdate{
				cancel:   true,
				clientID: clientID,
			}:
			case <-s.quit:
				return
			}
		},
	}

	select {
	case s.clientUpdates <- &clientUpdate{
		cancel:   false,
		clientID: clientID,
		client:   client,
	}:
	case <-s.quit:
		return nil, ErrServerShuttingDown
	}

	return client, nil
}
1019168
func (rcvr *Receiver) Subscribe(svcName string) {
	for _, subName := range rcvr.Subscriptions {
		if subName == svcName {
			return
		}
	}

	rcvr.Subscriptions = append(rcvr.Subscriptions, svcName)
}
592549
func (p *PubSub) Subscribe(id string, s Subscriber) error {
	p.m.Lock()
	defer p.m.Unlock()

	p.subscribers[id] = s
	return nil
}
661512
func NewSubscriptions(conn *Conn) *Subscriptions {
	s := &Subscriptions{
		channels:       make(map[string]struct{}),
		pchannels:      make(map[string]struct{}),
		conn:           conn,
		messageChannel: make(chan *Message, 100),
		ready:          true,
		readyChannel:   make(chan bool, 1),
		throwError:     false,
	}
	go s.receive()
	return s
}
1326350
func (c *ChromeTarget) Subscribe(method string, callback func(*ChromeTarget, []byte)) {
	c.eventLock.Lock()
	c.eventDispatcher[method] = callback
	c.eventLock.Unlock()
}
933546
func (c *Client) Subscribe(ctx context.Context, req *SubscriptionRequest) (string, error) {
	c.subscriptions.add(req)
	err := c.asynchronous.Send(ctx, req)
	if err != nil {
		// propagate send error
		return "", err
	}
	return req.SubID, nil
}
1224240
func (n *Stan) Subscribe(route string, handler stan.MsgHandler, opts ...stan.SubscriptionOption) error {
	subsc, err := n.Con.Subscribe(route, handler, opts...)
	if err != nil {
		return errors.New("Failed to make subcriptions for " + route + ": " + err.Error())
	}
	n.Opts.routes = append(n.Opts.routes, &Route{route: route, handler: handler, subsc: subsc})
	return nil
}
952527
func (c *TCP) Subscribe(tags []string) error {

	if len(tags) == 0 {
		return fmt.Errorf("Unable to subscribe - missing tags")
	}

	return c.encoder.Encode(&mist.Message{Command: "subscribe", Tags: tags})
}