// SubscriptionLoss returns true if the error indicates that the event subscription stream has been severed
// between mesos and a mesos client.
972510
func (e *Error) SubscriptionLoss() (result bool) {
	_, result = CodesIndicatingSubscriptionLoss[e.code]
	return
}
484838
func (sub *subscription) Dropped() (int, error) {
	sub.Lock()
	defer sub.Unlock()
	if sub.inboxSub == nil {
		return -1, ErrBadSubscription
	}
	return sub.inboxSub.Dropped()
}
639797
func (sub *subscription) ClearMaxPending() error {
	sub.Lock()
	defer sub.Unlock()
	if sub.inboxSub == nil {
		return ErrBadSubscription
	}
	return sub.inboxSub.ClearMaxPending()
}
408345
func (bc *brokerConsumer) subscriptionConsumer() {
	<-bc.wait // wait for our first piece of work

	for newSubscriptions := range bc.newSubscriptions {
		bc.updateSubscriptions(newSubscriptions)

		if len(bc.subscriptions) == 0 {
			// We're about to be shut down or we're about to receive more subscriptions.
			// Either way, the signal just hasn't propagated to our goroutine yet.
			<-bc.wait
			continue
		}

		response, err := bc.fetchNewMessages()

		if err != nil {
			Logger.Printf("consumer/broker/%d disconnecting due to error processing FetchRequest: %s\n", bc.broker.ID(), err)
			bc.abort(err)
			return
		}

		bc.acks.Add(len(bc.subscriptions))
		for child := range bc.subscriptions {
			child.feeder <- response
		}
		bc.acks.Wait()
		bc.handleResponses()
	}
}
944194
func (srv *Service) SubscribeAll(events map[string]EventSubscription) *SubscriptionError {
	for topic, regr := range events {
		if err := srv.Client.Subscribe(topic, regr.Handler, regr.Options); err != nil {
			return &SubscriptionError{
				Topic: topic,
				Inner: err,
			}
		}
	}

	return nil
}
1104032
func newSubscription(provider *SubscriptionProvider) *Subscription {
	return &Subscription{
		id:       uuid.NewV4(),
		events:   make(chan interface{}, subscriptionChanSize),
		quitChan: make(chan struct{}),
		provider: provider,
	}
}
191854
func (conn *Connection) Close() {
	subs := []string{}

	conn.subMu.Lock()

	for s := range conn.chanSubscriptions {
		subs = append(subs, s)
	}

	for s := range conn.subscriptions {
		subs = append(subs, s)
	}

	conn.subMu.Unlock()

	for _, s := range subs {
		err := conn.Unsubscribe(s)
		if err != nil {
			conn.logger.Warnf("Could not unsubscribe from %s: %s", s, err)
		}
	}


	conn.Flush()

	conn.conMu.Lock()
	defer conn.conMu.Unlock()

	conn.logger.Debug("Closing NATS connection")
	conn.nats.Close()
}
31265
func (s *DirectMessageService) EventsDestroy(id string) (*http.Response, error) {
	params := struct {
		ID string `url:"id,omitempty"`
	}{id}
	apiError := new(APIError)
	resp, err := s.sling.New().Delete("events/destroy.json").QueryStruct(params).Receive(nil, apiError)
	return resp, relevantError(err, *apiError)
}
343813
func DeleteSubscription(b bool) broker.Option {
	return func(o *broker.Options) {
		if o.Context == nil {
			o.Context = context.Background()
		}

		o.Context = context.WithValue(o.Context, deleteSubscription{}, b)
	}
}
20561
func (c *Client) LookupSubscription(subID string) (*SubscriptionRequest, error) {
	s, err := c.subscriptions.lookupBySubscriptionID(subID)
	if err != nil {
		return nil, err
	}
	return s.Request, nil
}
1229524
func (s *SubscriptionProvider) Close() (err error) {
	for _, sub := range s.subscriptions {
		serr := sub.Close()
		if serr != nil {
			err = serr
		}
	}

	return err
}
1003234
func (sub *ClientSubscription) Unsubscribe() {
	sub.quitWithError(nil, true)
	sub.errOnce.Do(func() { close(sub.err) })
}
1212947
func (n *natsStreamingRequester) Teardown() error {
	if err := n.sub.Unsubscribe(); err != nil {
		return err
	}
	n.sub = nil
	n.conn.Close()
	n.conn = nil
	return nil
}
1143896
func (p *pool) SubscriptionCount() int {
	p.RLock()
	defer p.RUnlock()
	return len(p.subs)
}
972821
func (lc *LightChain) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
	return lc.scope.Track(new(event.Feed).Subscribe(ch))
}
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
		}
	}
}
1189087
func isRemoteQSub(sub *subscription) bool {
	return sub != nil && sub.queue != nil && sub.client != nil && sub.client.kind == ROUTER
}