// TotalSubs returns total number of Subscriptions for this account.
492851
func (a *Account) TotalSubs() int {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return int(a.sl.Count())
}
1188399
func Subscribers() string {
	subs := make(map[string]bool) // no duplicates

	// get tags all clients subscribed to
	for i := range subscribers {
		s := subscribers[i].subscriptions.ToSlice()
		for j := range s {
			for k := range s[j] {
				subs[s[j][k]] = true
			}
		}
	}

	// slice it
	subSlice := []string{}
	for k, _ := range subs {
		subSlice = append(subSlice, k)
	}

	return strings.Join(subSlice, " ")
}
11971
func (s *Server) SubscribeUnbuffered(ctx context.Context, clientID string, query Query) (*Subscription, error) {
	return s.subscribe(ctx, clientID, query, 0)
}
463198
func (v *priorityClientPool) subscribeTotalCapacity(sub *tcSubscription) {
	v.lock.Lock()
	defer v.lock.Unlock()

	v.subs[sub] = struct{}{}
}
294665
func newSubscriber() *Subscriber {
	return &Subscriber{
		publish:  make(chan PubsubMessage),
		channels: map[string]struct{}{},
		patterns: map[string]*regexp.Regexp{},
	}
}
1021098
func (p *Peer) NumSubscribers() int {
	p.lock.RLock()
	subs := len(p.subscribers)
	p.lock.RUnlock()
	return subs
}
730710
func (m *Miniredis) PubSubNumPat() int {
	m.Lock()
	defer m.Unlock()

	return countPsubs(m.allSubscribers())
}
691167
func (w *WSEvents) redoSubscriptionsAfter(d time.Duration) {
	time.Sleep(d)

	for q := range w.subscriptions {
		err := w.ws.Subscribe(context.Background(), q)
		if err != nil {
			w.Logger.Error("Failed to resubscribe", "err", err)
		}
	}
}
326046
func NewSubscription(outCapacity int) *Subscription {
	return &Subscription{
		out:       make(chan Message, outCapacity),
		cancelled: make(chan struct{}),
	}
}
641212
func (r *Result) Total(t time.Time) time.Duration {
	return t.Sub(r.dnsStart)
}
861596
func (c Client) AddSubscription(s contract.Subscription) (string, error) {
	conn := c.Pool.Get()
	defer conn.Close()

	err := addSubscription(conn, &s)
	if err != nil {
		return "", err
	}

	return s.ID, nil
}
270741
func (s *Subscription) PendingLimits() (int, int, error) {
	if s == nil {
		return -1, -1, ErrBadSubscription
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.conn == nil {
		return -1, -1, ErrBadSubscription
	}
	if s.typ == ChanSubscription {
		return -1, -1, ErrTypeSubscription
	}
	return s.pMsgsLimit, s.pBytesLimit, nil
}
455355
func (s *ListsService) Subscribers(params *ListsSubscribersParams) (*Subscribers, *http.Response, error) {
	subscribers := new(Subscribers)
	apiError := new(APIError)
	resp, err := s.sling.New().Get("subscribers.json").QueryStruct(params).Receive(subscribers, apiError)
	return subscribers, resp, relevantError(err, *apiError)
}
529568
func TotalPages(totalRows uint, limit uint) uint {
	return uint(math.Ceil(float64(totalRows) / float64(limit)))
}
520390
func (cs *clientStore) getSubs(ID string) []*subState {
	cs.RLock()
	defer cs.RUnlock()
	c := cs.clients[ID]
	if c == nil {
		return nil
	}
	c.RLock()
	subs := c.getSubsCopy()
	c.RUnlock()
	return subs
}
871817
func (sub *Subscription) subUnsub(cmd string, channels ...string) error {
	err := sub.ensureProtocol()
	if err != nil {
		return err
	}
	pattern := false
	args := []interface{}{}
	for _, channel := range channels {
		if containsPattern(channel) {
			pattern = true
		}
		args = append(args, channel)
	}
	if pattern {
		cmd = "p" + cmd
	}
	err = sub.resp.sendCommand(cmd, args...)
	logCommand(cmd, args, err, sub.database.logging)
	return err
}
244611
func (p *Client) ClearSubscriptions() error {
	p.Lock()
	defer p.Unlock()
	zero := make([]byte, 2)
	for vh, s := range p.subs {
		if err := p.ac.Write(s.cccdh, zero); err != nil {
			return err
		}
		delete(p.subs, vh)
	}
	return nil
}