// Finds or creates a subscription
557500
func ensureSubscription(pubsubClient *pubsub.Client, topic *pubsub.Topic, subscriptionName string) *pubsub.Subscription {
	var subscription *pubsub.Subscription
	ctx, _ := context.WithTimeout(context.Background(), ContextDuration)
	subscription = pubsubClient.Subscription(subscriptionName)
	subscriptionExists, err := subscription.Exists(ctx)

	if err != nil {
		log.Fatalf("Could not check if subscription exists: %v", err)
	}
	if !subscriptionExists {
		new_subscription, err := pubsubClient.CreateSubscription(ctx, subscriptionName, topic, 0, nil)
		if err != nil {
			log.Fatalf("Could not create PubSub subscription: %v", err)
		}
		subscription = new_subscription
	}

	return subscription
}
1307539
func (m *SubscriptionManager) subscriptionHandler() {
	defer m.wg.Done()

	for {
		select {
		// A new subscription request has been received from a client.
		case msg := <-m.newSubscriptions:
			msg.errChan <- m.handleNewSubscription(msg)

		// A request to cancel an existing subscription has been
		// received from a client.
		case msg := <-m.cancelSubscriptions:
			m.handleCancelSubscription(msg)

		// A new block notification for the tip of the chain has been
		// received from the backing NotificationSource.
		case ntfn, ok := <-m.ntfnSource.Notifications():
			if !ok {
				log.Warn("Block source is unable to deliver " +
					"new updates")
				return
			}

			m.notifySubscribers(ntfn)

		case <-m.quit:
			return
		}
	}
}
266196
func (s *SubscriptionSet) Subscribe() (<-chan map[string]*UnitStatus, <-chan error) {
	// TODO: Make fully evented by using systemd 209 with properties changed values
	return s.conn.SubscribeUnitsCustom(time.Second, 0,
		mismatchUnitStatus,
		func(unit string) bool { return s.filter(unit) },
	)
}
226046
func (ss *RaftSubStore) CreateSub(sub *spb.SubState) error {
	gss := &ss.genericSubStore

	gss.Lock()
	defer gss.Unlock()

	// This store does not persist subscriptions, since it is done
	// in the actual RAFT log. This is just a wrapper to the streaming
	// sub store. We still need to apply limits.

	// If sub.ID is provided, check if already present, in which case
	// don't check limit.
	if sub.ID > 0 {
		if _, ok := gss.subs[sub.ID]; ok {
			return nil
		}
	}
	// Check limits
	if gss.limits.MaxSubscriptions > 0 && len(gss.subs) >= gss.limits.MaxSubscriptions {
		return ErrTooManySubs
	}

	// With new server, the sub.ID is set before this call is invoked,
	// and if that is the case, this is what we use. But let's support
	// not having one (in case we recover an existing store, or run a
	// mix of servers with different versions where the leader would
	// not be at a version that sets the sub.ID).
	if sub.ID == 0 {
		gss.maxSubID++
		sub.ID = gss.maxSubID
	} else if sub.ID > gss.maxSubID {
		gss.maxSubID = sub.ID
	}
	gss.subs[sub.ID] = emptySub

	return nil
}
780948
func subscribe(p *Proxy) {
	lumber.Trace("Adding proxy to subscribers...")

	mutex.Lock()
	subscribers[p.id] = p
	mutex.Unlock()
}
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
		}
	}
}
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
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}
93413
func (p *Provider) Subscribe(opts pubsub.HandlerOptions, h pubsub.MsgHandler) {
	logrus.Infof("Subscribing to %s successWriter/name %s", opts.Topic, opts.ServiceName+"."+opts.Name)
	logSuccess := logrus.WithField("pubsub", "kafka")
	successWriter := logSuccess.Writer()
	p.Reader = kafka.NewReader(kafka.ReaderConfig{
		Brokers:        p.Brokers,
		GroupID:        opts.ServiceName + "." + opts.Name,
		Topic:          opts.Topic,
		MaxBytes:       10e6, // 10MB
		CommitInterval: time.Second,
		MaxWait:        50 * time.Millisecond,
		Logger:         log.New(successWriter, "", log.Lshortfile),
		ErrorLogger:    log.New(ioutil.Discard, "", log.Lshortfile),
	})

	b := &backoff.Backoff{
		Min:    200 * time.Millisecond,
		Max:    600 * time.Second,
		Factor: 2,
		Jitter: true,
	}

	go func() {
		for {
			ctx := context.Background()
			m, err := p.Reader.FetchMessage(ctx)
			if err != nil {
				d := b.Duration()
				logrus.Errorf(
					"Subscription receive to topic %s failed, reconnecting in %v. Err: %v",
					opts.Topic, d, err,
				)
				time.Sleep(d)
			}

			b.Reset()

			msg := pubsub.Msg{
				ID:   string(m.Key),
				Data: m.Value,
				Ack: func() {
					p.Reader.CommitMessages(ctx, m)
				},
				Nack: func() {},
			}

			err = h(ctx, msg)
			if err != nil {
				break
			}

			if opts.AutoAck {
				msg.Ack()
			}

			logrus.Debugf("message at topic/partition/offset %v/%v/%v\n",
				m.Topic, m.Partition, m.Offset)
		}
		successWriter.Close()
	}()
}
919496
func (r *marathonClient) Subscribe(callback string) error {
	path := fmt.Sprintf("%s?callbackUrl=%s", marathonAPISubscription, callback)
	return r.apiPost(path, "", nil)

}
655826
func (p *Pubsub) resubscribe(write *writePump) {
	timer := gaugeLatency(PromReconnectLatency)
	PromReconnections.Inc()

	p.subsMu.Lock()
	defer p.subsMu.Unlock()

	for kind, recs := range p.subs {
		if recs == nil {
			continue
		}

		for _, ev := range recs.getList() {
			write.Data() <- command{
				command: EventType(kind).SubCommand(),
				channel: ev.name,
			}
		}
	}

	timer()
}
307274
func SubscriptionFromJSON(b []byte) (*Subscription, error) {
	var sub struct {
		Endpoint string
		Keys     struct {
			P256dh string
			Auth   string
		}
	}
	if err := json.Unmarshal(b, &sub); err != nil {
		return nil, err
	}

	b64 := base64.URLEncoding.WithPadding(base64.NoPadding)

	// Chrome < 52 incorrectly adds padding when Base64 encoding the values, so
	// we need to strip that out
	key, err := b64.DecodeString(strings.TrimRight(sub.Keys.P256dh, "="))
	if err != nil {
		return nil, err
	}

	auth, err := b64.DecodeString(strings.TrimRight(sub.Keys.Auth, "="))
	if err != nil {
		return nil, err
	}

	return &Subscription{sub.Endpoint, key, auth}, nil
}
87853
func (mc MongoClient) GetSubscriptions() ([]contract.Subscription, error) {
	return mc.getSubscriptions(bson.M{})
}
781665
func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- NewTxsEvent) event.Subscription {
	return pool.scope.Track(pool.txFeed.Subscribe(ch))
}
510394
func (c *PubSub) PSubscribe(patterns ...string) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	err := c.subscribe("psubscribe", patterns...)
	if c.patterns == nil {
		c.patterns = make(map[string]struct{})
	}
	for _, s := range patterns {
		c.patterns[s] = struct{}{}
	}
	return err
}
310648
func (r User_Customer) CreateNotificationSubscriber(keyName *string, resourceTableId *int) (resp bool, err error) {
	params := []interface{}{
		keyName,
		resourceTableId,
	}
	err = r.Session.DoRequest("SoftLayer_User_Customer", "createNotificationSubscriber", params, &r.Options, &resp)
	return
}
915666
func (r *serviceRegistry) subscription(service, name string) *callback {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.services[service].subscriptions[name]
}