Dynamically tune subscription only for the topic that dependencies need.
198085
@VisibleForTesting
  synchronized void consumerSubscriptionRebalance() {
    log.debug("Subscribed Topics " + this.consumer.subscription());
    if (!this.subscribedTopics.isEmpty()) {
      final Iterator<String> iter = this.subscribedTopics.iterator();
      final List<String> topics = new ArrayList<>();
      while (iter.hasNext()) {
        topics.add(iter.next());
      }
      this.subscribedTopics.clear();
      //re-subscribe topics that are needed
      this.consumer.subscribe(topics);
    }
  }
270314
function subscriptionInvoker (subscription, data, topic) {
    var result;

    if (subscription.fn.length === 3) {
      return new Promise(function promiseResolver (fulfill, reject) {
        subscription.fn.call(
          subscription.context, data, topic, function callback (err, succ) {
            return err ? reject(err) : fulfill(succ);
          }
        );
      });
    }

    try {
      result = subscription.fn.call(subscription.context, data, topic);
    } catch (e) {
      return Promise.reject(e);
    }

    if (result && typeof result.then === 'function') {
      return result;
    }

    return Promise.resolve(result);
  }
632013
function subscribe() {
  console.log('Subscribing client ' + clientId + ' to topic: ' + topics.subscribe);
  client.subscribe(topics.subscribe);
  isSubscribed = true;
}
1039295
function Subscription(name, topic) {
            this._pattern = "%s://%s.mns.%s.aliyuncs.com/topics/%s/subscriptions/%s";
            this._name = name;
            this._topic = topic;
            // make url
            this._urlAttr = this.makeAttrURL();
            // create the OpenStack object
            this._openStack = new AliMNS.OpenStack(topic.getAccount());
        }
1313365
function subscribe(topic, roomId) {
    return new Promise((resolve, reject) => {
        log_1.logger.info(`[subscribe] Preparing subscription: ${topic}: ${roomId}`);
        const subscription = exports.asteroid.subscribe(topic, roomId, true);
        exports.subscriptions.push(subscription);
        return subscription.ready
            .then((id) => {
            log_1.logger.info(`[subscribe] Stream ready: ${id}`);
            resolve(subscription);
        });
    });
}
340602
def subscribe(self, obj, handler):
    """"""
    if not isinstance(obj, LutronEntity):
      raise InvalidSubscription("Subscription target not a LutronEntity")
    _LOGGER.warning("DEPRECATED: Subscribing via Lutron.subscribe is obsolete. "
                    "Please use LutronEntity.subscribe")
    if obj not in self._legacy_subscribers:
      self._legacy_subscribers[obj] = handler
      obj.subscribe(self._dispatch_legacy_subscriber, None)
868988
def subscribe(hub, callback_url, topic_url, lease_seconds, secret,
              endpoint_hook_data):
    """"""

    for validate in hub.validators:
        error = validate(callback_url, topic_url, lease_seconds, secret,
                         endpoint_hook_data)
        if error:
            send_denied(hub, callback_url, topic_url, error)
            return

    if intent_verified(hub, callback_url, 'subscribe', topic_url,
                       lease_seconds):
        hub.storage[topic_url, callback_url] = {
            'lease_seconds': lease_seconds,
            'secret': secret,
        }
724469
function subscribe(topicArn) {

        // validate the topic arn
        if (!rxTopicArn.test(topicArn)) throw Error('Cannot subscribe to an invalid AWS Topic Arn: ' + topicArn);

        // if already subscribed then return now
        if (subscriptions[topicArn]) return subscriptions[topicArn].promise;

        // create and store the deferred promise
        const deferred = defer();
        subscriptions[topicArn] = deferred;

        // make the sns subscribe request
        const params = {
            Protocol: /^(https?):/.exec(config.endpoint)[1],
            TopicArn: topicArn,
            Endpoint: config.endpoint
        };
        config.sns.subscribe(params, function(err) {
            debug('Subscription request for ' + topicArn + (err ? ' failed : ' + err.message : ' sent.'));
            if (err) return deferred.reject(err);
        });

        return deferred.promise;
    }
903904
def start (self):
        '''
        

        '''
        self.sub = rospy.Subscriber(self.topic, Odometry, self.__callback)
269780
def sub(self, topic, channel):
        ''''''
        return self.send(' '.join((constants.SUB, topic, channel)))
861833
private function doBeforeCompile(): void
	{
		$builder = $this->getContainerBuilder();
		$dispatcher = $builder->getDefinition($this->prefix('dispatcher'));

		$subscribers = $builder->findByType(EventSubscriberInterface::class);
		foreach ($subscribers as $name => $subscriber) {
			$dispatcher->addSetup('addSubscriber', [$subscriber]);
		}
	}
1294783
def topic(self, channel, new_topic=None, tags=None):
        """"""
        params = [channel]
        if new_topic:
            params += new_topic
        self.send('TOPIC', params=params, source=self.nick, tags=tags)
138192
def _del_subscription(self, a_filter, session):
        
        deleted = 0
        try:
            subscriptions = self._subscriptions[a_filter]
            for index, (sub_session, qos) in enumerate(subscriptions):
                if sub_session.client_id == session.client_id:
                    self.logger.debug("Removing subscription on topic '%s' for client %s" %
                                      (a_filter, format_client_message(session=session)))
                    subscriptions.pop(index)
                    deleted += 1
                    break
        except KeyError:
            # Unsubscribe topic not found in current subscribed topics
            pass
        finally:
            return deleted
876805
def _dispatch_legacy_subscriber(self, obj, *args, **kwargs):
    """"""
    if obj in self._legacy_subscribers:
      self._legacy_subscribers[obj](obj)
237567
def unsubscribe_all(self):
        # type: () -> None
        """"""
        topics = list(self.topics.keys())
        for topic in topics:
            self.unsubscribe(topic)
946531
def start (self):
        '''
        

        '''
        self.sub = rospy.Subscriber(self.topic, BumperEvent, self.__callback)
1144035
def get_subscriptions(self, fetch=False):
        """"""
        return Subscriptions(
            self.resource.subscriptions, self.client, populate=fetch)