// This returns the structure responsible to keep track of
// pending messages and acks for a given subscription ID.
169081
func (ss *SQLSubStore) getOrCreateAcksPending(subid, seqno uint64) *sqlSubAcksPending {
	if !ss.cache.needsFlush {
		ss.cache.needsFlush = true
		ss.sqlStore.scheduleSubStoreFlush(ss)
	}
	ap := ss.cache.subs[subid]
	if ap == nil {
		ap = &sqlSubAcksPending{
			msgToRow: make(map[uint64]*sqlSubsPendingRow),
			ackToRow: make(map[uint64]*sqlSubsPendingRow),
			msgs:     make(map[uint64]struct{}),
			acks:     make(map[uint64]struct{}),
		}
		ss.cache.subs[subid] = ap
	}
	if seqno > ap.lastSent {
		ap.lastSent = seqno
	}
	return ap
}
743572
func (gd *GossipDelegate) NotifyJoin(node *memberlist.Node) {
	// Ignore self NotifyJoin
	nodeName := gd.parseMemberlistNodeName(node.Name)
	if nodeName == gd.nodeId {
		return
	}

	gd.updateGossipTs()

	// NotifyAlive should remove a node from memberlist if the
	// gossip version mismatches.
	// Nevertheless we are doing an extra check here.
	if err := gd.gossipChecks(node); err != nil {
		gd.RemoveNode(types.NodeId(nodeName))
	}
}
883040
func (ss *FileSubStore) AddSeqPending(subid, seqno uint64) error {
	ss.Lock()
	ss.updateSub.ID, ss.updateSub.Seqno = subid, seqno
	if err := ss.writeRecord(nil, subRecMsg, &ss.updateSub); err != nil {
		ss.Unlock()
		return err
	}
	si := ss.subs[subid]
	if si != nil {
		s := si.(*subscription)
		if seqno > s.sub.LastSent {
			s.sub.LastSent = seqno
		}
		s.seqnos[seqno] = struct{}{}
	}
	ss.Unlock()
	return nil
}
348446
func (r Network_Storage) GetNotificationSubscribers() (resp []datatypes.Notification_User_Subscriber, err error) {
	err = r.Session.DoRequest("SoftLayer_Network_Storage", "getNotificationSubscribers", nil, &r.Options, &resp)
	return
}
1095404
func (s *StanServer) processSub(c *channel, sr *pb.SubscriptionRequest, ackInbox string, subID uint64) (*subState, error) {
	// If channel not provided, we have to look it up
	var err error
	if c == nil {
		c, err = s.lookupOrCreateChannel(sr.Subject)
		if err != nil {
			s.log.Errorf("Unable to create channel for subscription on %q", sr.Subject)
			return nil, err
		}
	}
	var (
		sub *subState
		ss  = c.ss
	)
	// Will be true for durable queue subscribers and durable subscribers alike.
	isDurable := false
	// Will be set to false for en existing durable subscriber or existing
	// queue group (durable or not).
	setStartPos := true
	// Check for durable queue subscribers
	if sr.QGroup != "" {
		if sr.DurableName != "" {
			// For queue subscribers, we prevent DurableName to contain
			// the ':' character, since we use it for the compound name.
			if strings.Contains(sr.DurableName, ":") {
				s.log.Errorf("[Client:%s] Invalid DurableName (%q) for queue subscriber from %s",
					sr.ClientID, sr.DurableName, sr.Subject)
				return nil, ErrInvalidDurName
			}
			isDurable = true
			// Make the queue group a compound name between durable name and q group.
			sr.QGroup = fmt.Sprintf("%s:%s", sr.DurableName, sr.QGroup)
			// Clear DurableName from this subscriber.
			sr.DurableName = ""
		}
		// Lookup for an existing group. Only interested in situation where
		// the group exist, but is empty and had a shadow subscriber.
		ss.RLock()
		qs := ss.qsubs[sr.QGroup]
		if qs != nil {
			qs.Lock()
			if qs.shadow != nil {
				sub = qs.shadow
				qs.shadow = nil
				qs.subs = append(qs.subs, sub)
			}
			qs.Unlock()
			setStartPos = false
		}
		ss.RUnlock()
	} else if sr.DurableName != "" {
		// Check for DurableSubscriber status
		if sub = ss.LookupByDurable(durableKey(sr)); sub != nil {
			sub.RLock()
			clientID := sub.ClientID
			sub.RUnlock()
			if clientID != "" {
				s.log.Errorf("[Client:%s] Duplicate durable subscription registration", sr.ClientID)
				return nil, ErrDupDurable
			}
			setStartPos = false
		}
		isDurable = true
	}
	var (
		subStartTrace string
		subIsNew      bool
	)
	if sub != nil {
		// ok we have a remembered subscription
		sub.Lock()
		// Set ClientID and new AckInbox but leave LastSent to the
		// remembered value.
		sub.AckInbox = ackInbox
		sub.ClientID = sr.ClientID
		sub.Inbox = sr.Inbox
		sub.IsDurable = true
		// Use some of the new options, but ignore the ones regarding start position
		sub.MaxInFlight = sr.MaxInFlight
		sub.AckWaitInSecs = sr.AckWaitInSecs
		sub.ackWait = computeAckWait(sr.AckWaitInSecs)
		sub.stalled = false
		if len(sub.acksPending) > 0 {
			// We have a durable with pending messages, set newOnHold
			// until we have performed the initial redelivery.
			sub.newOnHold = true
			if !s.isClustered || s.isLeader() {
				s.setupAckTimer(sub, sub.ackWait)
			}
		}
		// Clear the IsClosed flags that were set during a Close()
		sub.IsClosed = false
		sub.Unlock()

		// Case of restarted durable subscriber, or first durable queue
		// subscriber re-joining a group that was left with pending messages.
		err = s.updateDurable(ss, sub)
	} else {
		subIsNew = true
		// Create sub here (can be plain, durable or queue subscriber)
		sub = &subState{
			SubState: spb.SubState{
				ClientID:      sr.ClientID,
				QGroup:        sr.QGroup,
				Inbox:         sr.Inbox,
				AckInbox:      ackInbox,
				MaxInFlight:   sr.MaxInFlight,
				AckWaitInSecs: sr.AckWaitInSecs,
				DurableName:   sr.DurableName,
				IsDurable:     isDurable,
			},
			subject:     sr.Subject,
			ackWait:     computeAckWait(sr.AckWaitInSecs),
			acksPending: make(map[uint64]int64),
			store:       c.store.Subs,
		}

		if setStartPos {
			// set the start sequence of the subscriber.
			var lastSent uint64
			subStartTrace, lastSent, err = s.setSubStartSequence(c, sr)
			if err == nil {
				sub.LastSent = lastSent
			}
		}

		if err == nil {
			// add the subscription to stan.
			// In cluster mode, the server decides of the subscription ID
			// (so that subscriptions have the same ID on replay). So
			// set it prior to this call.
			sub.ID = subID
			err = s.addSubscription(ss, sub)
			if err == nil && subID > 0 {
				ss.Lock()
				if subID >= c.nextSubID {
					c.nextSubID = subID + 1
				}
				ss.Unlock()
			}
		}
	}
	if err == nil && (!s.isClustered || s.isLeader()) {
		err = sub.startAckSub(s.nca, s.processAckMsg)
		if err == nil {
			// Need tp make sure that this subscription is processed by
			// NATS Server before sending response (since we use different
			// connection to send the response)
			s.nca.Flush()
		}
	}
	if err != nil {
		// Try to undo what has been done.
		s.closeMu.Lock()
		ss.Remove(c, sub, false)
		s.closeMu.Unlock()
		s.log.Errorf("Unable to add subscription for %s: %v", sr.Subject, err)
		return nil, err
	}
	if s.debug {
		traceCtx := subStateTraceCtx{clientID: sr.ClientID, isNew: subIsNew, startTrace: subStartTrace}
		traceSubState(s.log, sub, &traceCtx)
	}

	s.monMu.Lock()
	s.numSubs++
	s.monMu.Unlock()

	return sub, nil
}
563527
func (c *Connection) sendNotifications(msgID uint16, data []byte) {
	c.subscriptionsLock.RLock()
	defer c.subscriptionsLock.RUnlock()

	matched := false

	// send to notification to each subscriber
	for _, sub := range c.subscriptions[msgID] {
		log.WithFields(logger.Fields{
			"msg_name": sub.event.GetMessageName(),
			"msg_id":   msgID,
			"msg_size": len(data),
		}).Debug("Sending a notification to the subscription channel.")

		event := sub.msgFactory()
		if err := c.codec.DecodeMsg(data, event); err != nil {
			log.WithFields(logger.Fields{
				"msg_name": sub.event.GetMessageName(),
				"msg_id":   msgID,
				"msg_size": len(data),
			}).Errorf("Unable to decode the notification message: %v", err)
			continue
		}

		// send the message into the go channel of the subscription
		select {
		case sub.notifChan <- event:
			// message sent successfully
		default:
			// unable to write into the channel without blocking
			log.WithFields(logger.Fields{
				"msg_name": sub.event.GetMessageName(),
				"msg_id":   msgID,
				"msg_size": len(data),
			}).Warn("Unable to deliver the notification, reciever end not ready.")
		}

		matched = true
	}

	if !matched {
		log.WithFields(logger.Fields{
			"msg_id":   msgID,
			"msg_size": len(data),
		}).Info("No subscription found for the notification message.")
	}
}
504343
func (s *Subscription) GetSubscribed() bool {
	if s == nil || s.Subscribed == nil {
		return false
	}
	return *s.Subscribed
}
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
		}
	}
}
1267581
func (s *StanServer) processReplicatedSendAndAck(ssa *spb.SubSentAndAck) {
	c, err := s.lookupOrCreateChannel(ssa.Channel)
	if err != nil {
		return
	}
	sub := c.ss.LookupByAckInbox(ssa.AckInbox)
	if sub == nil {
		return
	}
	sub.Lock()
	defer sub.Unlock()

	// This is not optimized. The leader sent all accumulated sent and ack
	// sequences. For queue members, there is no much that can be done
	// because by nature seq will not be contiguous, but for non queue
	// subs, this could be optimized.
	for _, sequence := range ssa.Sent {
		// Update LastSent if applicable
		if sequence > sub.LastSent {
			sub.LastSent = sequence
		}
		// In case this is a queue member, update queue state's LastSent.
		if sub.qstate != nil && sequence > sub.qstate.lastSent {
			sub.qstate.lastSent = sequence
		}
		// Set 0 for expiration time. This will be computed
		// when the follower becomes leader and attempts to
		// redeliver messages.
		sub.acksPending[sequence] = 0
	}
	// Now remove the acks pending that we potentially just added ;-)
	for _, sequence := range ssa.Ack {
		delete(sub.acksPending, sequence)
	}
	// Don't set the sub.stalled here. Let that be done if the server
	// becomes leader and attempt the first deliveries.
}
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
}
781569
func (s *SubscriptionProvider) Notify(event interface{}) {
	s.RLock()
	defer s.RUnlock()
	for _, sub := range s.subscriptions {
		if err := sub.notify(event); err != nil {
			Log.Warnf("Failed notifying subscription (%s): %s", sub.id, err)
		}
	}
}
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()
	}
}
608803
func (s *StanServer) ackPublisher(iopm *ioPendingMsg) {
	msgAck := &iopm.pa
	msgAck.Guid = iopm.pm.Guid
	needed := msgAck.Size()
	s.tmpBuf = util.EnsureBufBigEnough(s.tmpBuf, needed)
	n, _ := msgAck.MarshalTo(s.tmpBuf)
	if s.trace {
		pm := &iopm.pm
		s.log.Tracef("[Client:%s] Acking Publisher subj=%s guid=%s", pm.ClientID, pm.Subject, pm.Guid)
	}
	s.ncs.Publish(iopm.m.Reply, s.tmpBuf[:n])
}
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
}
247186
func (s *Server) updateRouteSubscriptionMap(acc *Account, sub *subscription, delta int32) {
	if acc == nil || sub == nil {
		return
	}
	acc.mu.RLock()
	rm := acc.rm
	acc.mu.RUnlock()

	// This is non-nil when we know we are in cluster mode.
	if rm == nil {
		return
	}

	// We only store state on local subs for transmission across all other routes.
	if sub.client == nil || (sub.client.kind != CLIENT && sub.client.kind != SYSTEM && sub.client.kind != LEAF) {
		return
	}

	// Create the fast key which will use the subject or 'subject<spc>queue' for queue subscribers.
	var (
		_rkey  [1024]byte
		key    []byte
		update bool
	)
	if sub.queue != nil {
		// Just make the key subject spc group, e.g. 'foo bar'
		key = _rkey[:0]
		key = append(key, sub.subject...)
		key = append(key, byte(' '))
		key = append(key, sub.queue...)
		// We always update for a queue subscriber since we need to send our relative weight.
		update = true
	} else {
		key = sub.subject
	}

	// Copy to hold outside acc lock.
	var n int32
	var ok bool

	acc.mu.Lock()
	if n, ok = rm[string(key)]; ok {
		n += delta
		if n <= 0 {
			delete(rm, string(key))
			update = true // Update for deleting (N->0)
		} else {
			rm[string(key)] = n
		}
	} else if delta > 0 {
		n = delta
		rm[string(key)] = delta
		update = true // Adding a new entry for normal sub means update (0->1)
	}
	acc.mu.Unlock()

	if !update {
		return
	}
	// We need to send out this update.

	// If we are sending a queue sub, copy and place in the queue weight.
	if sub.queue != nil {
		sub.client.mu.Lock()
		nsub := *sub
		sub.client.mu.Unlock()
		nsub.qw = n
		sub = &nsub
	}

	// Note that queue unsubs where entry.n > 0 are still
	// subscribes with a smaller weight.
	if n > 0 {
		s.broadcastSubscribe(sub)
	} else {
		s.broadcastUnSubscribe(sub)
	}
}
15116
func (c *cache) notify(id types.UID, timestamp time.Time) {
	list, ok := c.subscribers[id]
	if !ok {
		// No one to notify.
		return
	}
	newList := []*subRecord{}
	for i, r := range list {
		if timestamp.Before(r.time) {
			// Doesn't meet the time requirement; keep the record.
			newList = append(newList, list[i])
			continue
		}
		r.ch <- c.get(id)
		close(r.ch)
	}
	if len(newList) == 0 {
		delete(c.subscribers, id)
	} else {
		c.subscribers[id] = newList
	}
}
591899
func (m *SubscriptionManager) notifySubscriber(sub *newSubscription,
	block BlockNtfn) {

	select {
	case sub.ntfnQueue.ChanIn() <- block:
	case <-sub.quit:
	case <-m.quit:
		return
	}
}