// Subscribe creates a subscription for events of the given types. The
// subscription's channel is closed when it is unsubscribed
// or the mux is closed.
167024
func (mux *TypeMux) Subscribe(types ...interface{}) *TypeMuxSubscription {
	sub := newsub(mux)
	mux.mutex.Lock()
	defer mux.mutex.Unlock()
	if mux.stopped {
		// set the status to closed so that calling Unsubscribe after this
		// call will short circuit.
		sub.closed = true
		close(sub.postC)
	} else {
		if mux.subm == nil {
			mux.subm = make(map[reflect.Type][]*TypeMuxSubscription)
		}
		for _, t := range types {
			rtyp := reflect.TypeOf(t)
			oldsubs := mux.subm[rtyp]
			if find(oldsubs, sub) != -1 {
				panic(fmt.Sprintf("event: duplicate type %s in Subscribe", rtyp))
			}
			subs := make([]*TypeMuxSubscription, len(oldsubs)+1)
			copy(subs, oldsubs)
			subs[len(oldsubs)] = sub
			mux.subm[rtyp] = subs
		}
	}
	return sub
}
179212
func (em *clientEventMonitor) Subscribe(ID string) (*Subscription, error) {
	em.Lock()
	defer em.Unlock()

	s := &Subscription{
		ID:            ID,
		cancelChannel: make(chan chan struct{}),
		eventChannel:  make(chan *dockerclient.APIEvents),
		monitorDone:   em.done,
		handlers:      make(map[string]HandlerFunc),
		monitor:       em,
		lock:          sync.RWMutex{},
	}

	em.subscriptions[ID] = append(em.subscriptions[ID], s)
	s.run()

	return s, nil
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}
611859
func (o *os) Subscribe(ctx context.Context, h Handler, types ...string) error {
	req := &ev.StreamRequest{}
	for _, typ := range types {
		req.Types = append(req.Types, typ)
	}

	stream, err := o.cl.Stream(ctx, req)
	if err != nil {
		return err
	}

	for {
		rsp, err := stream.Recv()
		if err != nil {
			return err
		}
		h(toRecord(rsp.Record))
	}

	return nil
}
192584
func (am *Manager) Subscribe(sink chan<- WalletEvent) event.Subscription {
	return am.feed.Subscribe(sink)
}
1322813
func (em *EventMeter) subscribe() error {
	for query := range em.queryToMetricMap {
		if err := em.wsc.Subscribe(context.TODO(), query); err != nil {
			return err
		}
	}
	return nil
}
1261518
func (e *EventListener) Subscribe(appID string, objectTypes []string, objectID string) (*EventSubscriber, error) {
	e.subMtx.Lock()
	defer e.subMtx.Unlock()
	if e.IsClosed() {
		return nil, errors.New("event listener closed")
	}
	s := &EventSubscriber{
		Events:      make(chan *ct.Event),
		l:           e,
		queue:       make(chan *ct.Event, eventBufferSize),
		stop:        make(chan struct{}),
		appID:       appID,
		objectTypes: objectTypes,
		objectID:    objectID,
	}
	go s.loop()
	if _, ok := e.subscribers[appID]; !ok {
		e.subscribers[appID] = make(map[*EventSubscriber]struct{})
	}
	e.subscribers[appID][s] = struct{}{}
	return s, nil
}
1373360
func Subscribe(info *mesos.FrameworkInfo) *scheduler.Call {
	return &scheduler.Call{
		Type:        scheduler.Call_SUBSCRIBE,
		FrameworkID: info.GetID(),
		Subscribe:   &scheduler.Call_Subscribe{FrameworkInfo: info},
	}
}
567520
func (cli *Client) Subscribe(opts *SubscribeOptions) error {
	// Lock for reading and updating.
	cli.muConn.Lock()

	// Unlock.
	defer cli.muConn.Unlock()

	// Check the Network Connection.
	if cli.conn == nil {
		return ErrNotYetConnected
	}

	// Check the existence of the options.
	if opts == nil || len(opts.SubReqs) == 0 {
		return packet.ErrInvalidNoSubReq
	}

	// Define a Packet Identifier.
	var packetID uint16

	// Define an error.
	var err error

	// Lock for updating the Session.
	cli.muSess.Lock()

	defer cli.muSess.Unlock()

	// Generate a Packet Identifer.
	if packetID, err = cli.generatePacketID(); err != nil {
		return err
	}

	// Create subscription requests for the SUBSCRIBE Packet.
	var subReqs []*packet.SubReq

	for _, s := range opts.SubReqs {
		subReqs = append(subReqs, &packet.SubReq{
			TopicFilter: s.TopicFilter,
			QoS:         s.QoS,
		})
	}

	// Create a SUBSCRIBE Packet.
	p, err := packet.NewSUBSCRIBE(&packet.SUBSCRIBEOptions{
		PacketID: packetID,
		SubReqs:  subReqs,
	})
	if err != nil {
		return err
	}

	// Set the Packet to the Session.
	cli.sess.sendingPackets[packetID] = p

	// Set the subscription information to
	// the Network Connection.
	for _, s := range opts.SubReqs {
		cli.conn.unackSubs[string(s.TopicFilter)] = s.Handler
	}

	// Send the Packet to the Server.
	cli.conn.send <- p

	return nil
}
96004
func (b *Buffer) subscribe(msgc chan<- *rfc5424.Message, donec <-chan struct{}) {
	b.subs[msgc] = struct{}{}

	go func() {
		select {
		case <-donec:
		case <-b.donec:
		}

		b.mu.Lock()
		defer b.mu.Unlock()

		delete(b.subs, msgc)
		close(msgc)
	}()
}
1062733
func (sc *conn) Subscribe(subject string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
	return sc.subscribe(subject, "", cb, options...)
}
484999
func (ps *PubSub) Subscribe(topic string, ttl time.Duration) Subscription {
	sub := &subscription{
		top: topic,
		ttl: ttl,
		c:   make(chan interface{}, subscriptionBuffSize),
	}

	ps.Lock()
	// Add subscription to subscriptions map
	s, exists := ps.subscriptions[topic]
	// If no subscription set for the topic exists, create one
	if !exists {
		s = NewSet()
		ps.subscriptions[topic] = s
	}
	ps.Unlock()

	// Add the subscription
	s.Add(sub)

	// When the timeout expires, remove the subscription
	time.AfterFunc(ttl, func() {
		ps.unSubscribe(sub)
	})
	return sub
}
731710
func (_m *Light) Subscribe() *common.Subscription {
	ret := _m.Called()

	var r0 *common.Subscription
	if rf, ok := ret.Get(0).(func() *common.Subscription); ok {
		r0 = rf()
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*common.Subscription)
		}
	}

	return r0
}
611085
func (m *Monitor) Subscribe() chan runc.Exit {
	c := make(chan runc.Exit, bufferSize)
	m.Lock()
	m.subscribers[c] = struct{}{}
	m.Unlock()
	return c
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
592549
func (p *PubSub) Subscribe(id string, s Subscriber) error {
	p.m.Lock()
	defer p.m.Unlock()

	p.subscribers[id] = s
	return nil
}
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
}