// triggerPushSubscriptions is used internally for starting iterations
// on Push subscriptions. Whenever new item is added to the push index,
// this function should be called.
960667
func (db *DB) triggerPushSubscriptions() {
	db.pushTriggersMu.RLock()
	triggers := db.pushTriggers
	db.pushTriggersMu.RUnlock()

	for _, t := range triggers {
		select {
		case t <- struct{}{}:
		default:
		}
	}
}
1364724
func (s *SetIdentityPoolConfigurationInput) SetPushSync(v *PushSync) *SetIdentityPoolConfigurationInput {
	s.PushSync = v
	return s
}
766482
func (r User_Customer_MobileDevice) GetPushNotificationSubscriptions() (resp []datatypes.Notification_User_Subscriber, err error) {
	err = r.Session.DoRequest("SoftLayer_User_Customer_MobileDevice", "getPushNotificationSubscriptions", nil, &r.Options, &resp)
	return
}
1104032
func newSubscription(provider *SubscriptionProvider) *Subscription {
	return &Subscription{
		id:       uuid.NewV4(),
		events:   make(chan interface{}, subscriptionChanSize),
		quitChan: make(chan struct{}),
		provider: provider,
	}
}
709569
func (s *Server) ringSubscribersNotify() {
	for change := range s.subsChangeChan {
		s.ringSubs.RLock()
		ring := &pb.Ring{
			Ring:    *change.rb,
			Version: change.v,
		}
		for id, ch := range s.ringSubs.subs {
			go func(id string, ch chan *pb.Ring, ring *pb.Ring) {
				ch <- ring
			}(id, ch, ring)
		}
		s.ringSubs.RUnlock()
	}
}
666261
func (s *DescribeHsmOutput) SetSubscriptionStartDate(v string) *DescribeHsmOutput {
	s.SubscriptionStartDate = &v
	return s
}
612662
func (n *Notifier) takeSubscription() *Subscription {
	n.mu.Lock()
	defer n.mu.Unlock()
	n.callReturned = true
	return n.sub
}
1036491
func (s *ListSubscriptionsOutput) SetSubscriptions(v []*Subscription) *ListSubscriptionsOutput {
	s.Subscriptions = v
	return s
}
513019
func (c *Ring) PSubscribe(channels ...string) *PubSub {
	if len(channels) == 0 {
		panic("at least one channel is required")
	}

	shard, err := c.shards.GetByKey(channels[0])
	if err != nil {
		// TODO: return PubSub with sticky error
		panic(err)
	}
	return shard.Client.PSubscribe(channels...)
}
344744
func (fs *FS) SubscribeToEvents(ch chan<- FSEvent) {
	fs.eventsLock.Lock()
	defer fs.eventsLock.Unlock()
	fs.events[ch] = true
}
1166021
func NewPushNodesByBackendIDsToFrontendArgs(backendNodeIDs []BackendNodeID) *PushNodesByBackendIDsToFrontendArgs {
	args := new(PushNodesByBackendIDsToFrontendArgs)
	args.BackendNodeIDs = backendNodeIDs
	return args
}
1021098
func (p *Peer) NumSubscribers() int {
	p.lock.RLock()
	subs := len(p.subscribers)
	p.lock.RUnlock()
	return subs
}
1145583
func pubsubPushHandler(c *router.Context) {
	rc := requestContext(*c)
	body, err := ioutil.ReadAll(rc.Request.Body)
	if err != nil {
		rc.fail(500, "Failed to read the request: %s", err)
		return
	}
	if err = globalEngine.ProcessPubSubPush(rc.Context, body); err != nil {
		rc.err(err, "Failed to process incoming PubSub push")
		return
	}
	rc.ok()
}
628157
func (s *SubscriptionProvider) Subscribe() *Subscription {
	s.Lock()
	defer s.Unlock()
	if s.subscriptions == nil {
		s.subscriptions = make(map[string]*Subscription)
	}
	sub := newSubscription(s)
	s.subscriptions[sub.id.String()] = sub

	return sub
}
1220624
func isSubscriptionType(t reflect.Type) bool {
	for t.Kind() == reflect.Ptr {
		t = t.Elem()
	}
	return t == subscriptionType
}
898702
func (i *InvoiceRegistry) notifyHodlSubscribers(hodlEvent HodlEvent) {
	subscribers, ok := i.hodlSubscriptions[hodlEvent.Hash]
	if !ok {
		return
	}

	// Notify all interested subscribers and remove subscription from both
	// maps. The subscription can be removed as there only ever will be a
	// single resolution for each hash.
	for subscriber := range subscribers {
		select {
		case subscriber <- hodlEvent:
		case <-i.quit:
			return
		}

		delete(i.hodlReverseSubscriptions[subscriber], hodlEvent.Hash)
	}

	delete(i.hodlSubscriptions, hodlEvent.Hash)
}
988216
func (m *SubscriptionManager) handleNewSubscription(sub *newSubscription) error {
	log.Infof("Registering block subscription: id=%d", sub.id)

	// We'll start by retrieving a backlog of notifications from the
	// client's best height.
	blocks, currentHeight, err := m.ntfnSource.NotificationsSinceHeight(
		sub.bestHeight,
	)
	if err != nil {
		return fmt.Errorf("unable to retrieve blocks since height=%d: "+
			"%v", sub.bestHeight, err)
	}

	// We'll then attempt to deliver these notifications.
	log.Debugf("Delivering backlog of block notifications: id=%d, "+
		"start_height=%d, end_height=%d", sub.id, sub.bestHeight,
		currentHeight)

	for _, block := range blocks {
		m.notifySubscriber(sub, block)
	}

	// With the notifications delivered, we can keep track of the new client
	// internally in order to deliver new block notifications about the
	// chain.
	m.subscribers[sub.id] = sub

	return nil
}