Sets up this Node with the specified Interfaces before it is run.

		Args:
			puller: Indication if a Puller Interface should be created.
			subscriptions: Collection of the Subscriber Interfaces to be created and their Slots.
504765
def setup(self, puller: bool=None, subscriptions: Dict[str, Any]={}):
		
		if puller:
			puller = self._zmq.socket(zmq.PULL)
			ip, port, host = self.rslv('rcv')
			puller.bind('tcp://{}:{}'.format(host or ip, port))
			self.poll(puller)
		if subscriptions:
			for publisher in subscriptions:  # type: str
				self.add(publisher, subscriptions[publisher].get('slots'), subscriptions[publisher].get('buffer-length'))
			logger.info('Listening to %s', {
				k: (1 if subscriptions[k].get('slots') is None else len(subscriptions[k].get('slots')))
				for k in subscriptions
			})
645409
def create_subscription(subscription_params = {})
      created_at = Time.current
      if subscription_params[:subscribing] == false && subscription_params[:subscribing_to_email].nil?
        subscription_params[:subscribing_to_email] = subscription_params[:subscribing] 
      end
      subscription = subscriptions.new(subscription_params)
      subscription.subscribing? ?
        subscription.assign_attributes(subscribing: true, subscribed_at: created_at) :
        subscription.assign_attributes(subscribing: false, unsubscribed_at: created_at)
      subscription.subscribing_to_email? ?
        subscription.assign_attributes(subscribing_to_email: true, subscribed_to_email_at: created_at) :
        subscription.assign_attributes(subscribing_to_email: false, unsubscribed_to_email_at: created_at)
      subscription.optional_targets = (subscription.optional_targets || {}).with_indifferent_access
      optional_targets = {}.with_indifferent_access
      subscription.optional_target_names.each do |optional_target_name|
        optional_targets = subscription.subscribing_to_optional_target?(optional_target_name) ?
          optional_targets.merge(
            Subscription.to_optional_target_key(optional_target_name) => true,
            Subscription.to_optional_target_subscribed_at_key(optional_target_name) => created_at
          ) :
          optional_targets.merge(
            Subscription.to_optional_target_key(optional_target_name) => false,
            Subscription.to_optional_target_unsubscribed_at_key(optional_target_name) => created_at
          )
      end
      subscription.assign_attributes(optional_targets: optional_targets)
      subscription.save ? subscription : nil
    end
837939
public void subscribe(ArrayList<Long> tokens) {
        if(ws != null) {
            if (ws.isOpen()) {
                createTickerJsonObject(tokens, mSubscribe);
                ws.sendText(createTickerJsonObject(tokens, mSubscribe).toString());
                subscribedTokens.addAll(tokens);
                for(int i = 0; i < tokens.size(); i++){
                    modeMap.put(tokens.get(i), modeQuote);
                }
            }else {
                if(onErrorListener != null) {
                    onErrorListener.onError(new KiteException("ticker is not connected", 504));
                }
            }
        }else {
            if(onErrorListener != null) {
                onErrorListener.onError(new KiteException("ticker is null not connected", 504));
            }
        }
    }
879458
def subscribe(self, destination, id, ack='auto', headers=None, **keyword_headers):
        
        self.transport.subscriptions[id] = destination
574729
public void init(IPersistentSubscriptionStore storageService) {
        Log.debug("init invoked");

        m_storageService = storageService;

        //reload any subscriptions persisted
        if (Log.DEBUG) {
            Log.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree());
        }
        
        for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
            Log.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic());
            addDirect(subscription);
        }
        if (Log.DEBUG) {
            Log.debug("Finished loading. Subscription tree after {}", dumpTree());
        }
    }
545283
@Override
    public void init(CollectorManager collectorManager) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
            Tr.event(tc, "Subscribing to sources " + this, taskMap.keySet());
        }
        try {
            this.collectorMgr = collectorManager;
            //Get the source Ids from the task map and subscribe to relevant sources
            collectorMgr.subscribe(this, new ArrayList<String>(taskMap.keySet()));
        } catch (Exception e) {

        } finally {
            latch.countDown();
        }
    }
515782
def select_subscription(subs_code, subscriptions):
    
    if subs_code and subscriptions:
        for subs in subscriptions:
            if (subs.subscription_code == subs_code):
                return subs
    return None
37330
def subscribe(self):
        '''  '''
        try:
            LOG.debug('Creating Kinesis subscription')
            if self.starting_position_ts:
                self._lambda_client \
                    .create_event_source_mapping(
                        EventSourceArn=self.stream_name,
                        FunctionName=self.function_name,
                        BatchSize=self.batch_size,
                        StartingPosition=self.starting_position,
                        StartingPositionTimestamp=self.starting_position_ts)
            else:
                self._lambda_client \
                    .create_event_source_mapping(
                        EventSourceArn=self.stream_name,
                        FunctionName=self.function_name,
                        BatchSize=self.batch_size,
                        StartingPosition=self.starting_position)
            LOG.debug('Subscription created')
        except botocore.exceptions.ClientError as ex:
            response_code = ex.response['Error']['Code']
            if response_code == 'ResourceConflictException':
                LOG.debug('Subscription exists. Updating ...')
                resp = self._lambda_client\
                           .list_event_source_mappings(
                                FunctionName=self.function_name,
                                EventSourceArn=self.stream_name)
                uuid = resp['EventSourceMappings'][0]['UUID']
                self._lambda_client \
                    .update_event_source_mapping(
                        UUID=uuid,
                        FunctionName=self.function_name,
                        Enabled=True,
                        BatchSize=self.batch_size)
            else:
                LOG.error('Subscription failed, error=%s' % str(ex))
                raise ex
619507
public void setSubscribers(java.util.Collection<Subscriber> subscribers) {
        if (subscribers == null) {
            this.subscribers = null;
            return;
        }

        this.subscribers = new java.util.ArrayList<Subscriber>(subscribers);
    }
1111609
function subscribeOn(scheduler, delay) {
    if (delay === void 0) {
        delay = 0;
    }
    return function subscribeOnOperatorFunction(source) {
        return source.lift(new SubscribeOnOperator(scheduler, delay));
    };
}
1279109
public void subscribe(Class<? extends Event> eventType, Class<? extends Subscriber> subscriberType) {
        if (!subscriptionMap.containsKey(eventType)) {
            subscriptionMap.put(eventType, new ArrayList<>());
        }
        final ArrayList<Class<? extends Subscriber>> subscribers = subscriptionMap.get(eventType);
        if (!subscribers.contains(subscriberType)) {
            subscribers.add(subscriberType);
        }
    }
460693
async def _set_subscriptions(self, subscriptions):
        

        url, params = self._get_subscriptions_endpoint()

        data = {
            'object': 'page',
            'callback_url': self.webhook_url,
            'fields': ', '.join(subscriptions),
            'verify_token': self.verify_token,
        }

        headers = {
            'Content-Type': 'application/json',
        }

        post = self.session.post(
            url,
            params=params,
            data=ujson.dumps(data),
            headers=headers,
        )

        async with post as r:
            await self._handle_fb_response(r)
            data = await r.json()
662558
public IPromise<KSubscription> _subscribe( Callback subscriber ) {
        if ( subscribers == null )
            subscribers = new HashMap<>();
        int id = subsIdCount++;
        KSubscription subs = new KSubscription(self(), id);
        subscribers.put( id, new SubscriberEntry(id, subs, subscriber) );
        return new Promise<>(subs);
    }
111248
def _add_subscribers_for_type(self, callback_type, subscribers, callbacks, **kwargs):
        '''  '''
        for subscriber in subscribers:
            callback_name = 'on_' + callback_type
            if hasattr(subscriber, callback_name):
                _function = functools.partial(getattr(subscriber, callback_name), **kwargs)
                callbacks.append(_function)
368573
public function addSubscriber(SubscriberInterface $subscriber): NodeInterface
    {
        $this->_subscribers[$subscriber->getId()] = $subscriber;

        return $this;
    }
1090449
public Subscription addSubscriber(Function<? super E, Subscription> subscriber) {
        Objects.requireNonNull(subscriber);
        subscribers.add(subscriber);

        List<E> keys = new ArrayList<>(map.keySet());
        keys.forEach(key -> {
            List<Subscription> otherSubs = map.get(key);
            Subscription sub = subscriber.apply(key);
            otherSubs.add(sub);
            map.put(key, otherSubs);
        });

        return () -> removeSubscriber(subscriber);
    }
888435
static <T> Subscriber<T> wrap(
            Subscriber<T> downstream, List<RunnableInstrumenter> instrumentations) {
        if (downstream instanceof FlowableSubscriber) {
            return new RxInstrumentedFlowableSubscriber<>(downstream, instrumentations);
        }
        return new RxInstrumentedSubscriber<>(downstream, instrumentations);
    }