Subcribe to event
@param  string $id Event ID
@param  object $obj Subscriber
@param  callable $cb Callback
@return boolean       Success
960713
public function sub($id, $obj, $cb)
    {
        if (!isset($this->events[$id])) {
            return false;
        }
        return $this->events[$id]->sub($obj, $cb);
    }
25942
public function addSubscription( $email, $user, $contentID, $languageID, $subscriptionType, $currentTime, $activate = true )
    {
        //1. insert into subscriber
        $ezcommentsINI = eZINI::instance( 'ezcomments.ini' );
        $subscriber = ezcomSubscriber::fetchByEmail( $email );
        //if there is no data in subscriber for same email, save it
        if ( is_null( $subscriber ) )
        {
            $subscriber = ezcomSubscriber::create();
            $subscriber->setAttribute( 'user_id', $user->attribute( 'contentobject_id' ) );
            $subscriber->setAttribute( 'email', $email );
            if ( $user->isAnonymous() )
            {
                $util = ezcomUtility::instance();
                $hashString = $util->generateSusbcriberHashString( $subscriber );
                $subscriber->setAttribute( 'hash_string', $hashString );
            }
            $subscriber->store();
            eZDebugSetting::writeNotice( 'extension-ezcomments', 'Subscriber does not exist, added one', __METHOD__ );
            $subscriber = ezcomSubscriber::fetchByEmail( $email );
        }
        else
        {
            if ( $subscriber->attribute( 'enabled' ) == false )
            {
                throw new Exception('Subscription can not be added because the subscriber is disabled.', self::ERROR_SUBSCRIBER_DISABLED );
            }
        }
        //3 insert into subscription table
        // if there is no data in ezcomment_subscription with given contentobject_id and subscriber_id
        $hasSubscription = ezcomSubscription::exists( $contentID,
                                                      $languageID,
                                                      $subscriptionType,
                                                      $email );
        if ( $hasSubscription === false )
        {
            $subscription = ezcomSubscription::create();
            $subscription->setAttribute( 'user_id', $user->attribute( 'contentobject_id' ) );
            $subscription->setAttribute( 'subscriber_id', $subscriber->attribute( 'id' ) );
            $subscription->setAttribute( 'subscription_type', $subscriptionType );
            $subscription->setAttribute( 'content_id', $contentID );
            $subscription->setAttribute( 'language_id', $languageID );
            $subscription->setAttribute( 'subscription_time', $currentTime );
            $defaultActivated = $ezcommentsINI->variable( 'CommentSettings', 'SubscriptionActivated' );

            if ( $user->isAnonymous() && $defaultActivated !== 'true' && $activate === true )
            {
                $subscription->setAttribute( 'enabled', 0 );
                $utility = ezcomUtility::instance();
                $subscription->setAttribute( 'hash_string', $utility->generateSubscriptionHashString( $subscription ) );
                $subscription->store();

                $result = ezcomSubscriptionManager::sendActivationEmail( eZContentObject::fetch( $contentID ),
                                                                         $subscriber,
                                                                         $subscription );
                if ( !$result )
                {
                    eZDebug::writeError( "Error sending mail to '$email'", __METHOD__ );
                }
            }
            else
            {
                $subscription->setAttribute( 'enabled', 1 );
                $subscription->store();
            }
            eZDebugSetting::writeNotice( 'extension-ezcomments', 'No existing subscription for this content and user, added one', __METHOD__ );
        }
    }
1015971
protected void subscribe( String id, MessagingContext ctx ) throws IOException {

		if( ! canProceed())
			return;

		Set<MessagingContext> sub = this.routingContext.subscriptions.get( id );
		if( sub == null ) {
			sub = new HashSet<> ();
			this.routingContext.subscriptions.put( id, sub );
		}

		sub.add( ctx );
	}
1188036
public function subscribe(ConnectionInterface $connection, stdClass $payload)
    {
        $this->verifySignature($connection, $payload);

        $this->saveConnection($connection);

        $channelData = json_decode($payload->channel_data);
        $this->users[$connection->socketId] = $channelData;

        // Send the success event
        $connection->send(json_encode([
            'event' => 'pusher_internal:subscription_succeeded',
            'channel' => $this->channelName,
            'data' => json_encode($this->getChannelData()),
        ]));

        $this->broadcastToOthers($connection, [
            'event' => 'pusher_internal:member_added',
            'channel' => $this->channelName,
            'data' => json_encode($channelData),
        ]);
    }
274565
public function getUniqueId($cb, $id = null)
    {
        $this->onResponse->push($cb);
        if ($id) {
            $this->fetch('UID', $id, null, false, self::TAG_GETUID);
        } else {
            $this->fetch('UID', 1, INF, false, self::TAG_GETUID);
        }
    }
61737
public function addListener($cb)
    {
        if ($this->state === self::STATE_DONE) {
            if ($cb !== null) {
                $cb($this);
            }
            return;
        }
        if ($cb !== null) {
            $this->listeners->push($cb);
        }
        if ($this->state === self::STATE_WAITING) {
            $i = 1;
            $n = func_num_args();
            while ($i < $n) {
                $this->args[] = func_get_arg($i);
                ++$i;
            }
            $this->state = self::STATE_RUNNING;
            $func = $this->producer;
            $func($this);
        }
    }
534571
public function objectIdChanged(string $objectType, string $oldObjectId, string $newObjectId)
    {
        //==============================================================================
        // Create Event Object
        $event = new ObjectsIdChangedEvent($this->getWebserviceId(), $objectType, $oldObjectId, $newObjectId);
        //==============================================================================
        //      Dispatch Event
        $this->getEventDispatcher()->dispatch(ObjectsIdChangedEvent::NAME, $event);

        return true;
    }
870758
public function unsubFromAll($obj)
    {
        foreach ($this->events as $event) {
            $event->unsub($obj);
        }
        return true;
    }
769531
public void set_async_reply_cb(final int id, final CallBack cb) {
        final AsyncCallObject aco = async_request_table.get(id);
        if (aco != null) {
            aco.cb = cb;
        }
    }
245649
public function addEvent($id, callable $cb)
    {
        if (!isset($this->events[$id])) {
            $this->events[$id] = [];
        }

        $this->events[$id][] = $cb;
    }
1253657
public function subscribe($name, $id = null, $callback = null, $timeout = null, $failswitch = false) {
        $self = $this;
        if (!is_string($name)) {
            throw new TypeError('topic name must be a string');
        }
        if (is_callable($id) && !is_callable($callback)) {
            $timeout = $callback;
            $callback = $id;
            $id = null;
        }
        if (!is_callable($callback)) {
            throw new TypeError('callback must be a function.');
        }
        if (!isset($this->topics[$name])) {
            $this->topics[$name] = array();
        }
        if ($id === null) {
            if ($this->id == null) {
                $this->id = $this->autoId();
            }
            $this->id->then(function($id) use ($self, $name, $callback, $timeout, $failswitch) {
                $self->;
            });
            return;
        }
        // Default subscribe timeout is 5 minutes.
        if (!is_int($timeout)) $timeout = 300000;
        $topic = $this->getTopic($name, $id);
        if ($topic === null) {
            $topic = new stdClass();
            $settings = new InvokeSettings(array(
                'idempotent' => true,
                'failswitch' => $failswitch,
                'timeout' => $timeout
            ));
            $cb = function() use ($self, &$cb, $topic, $name, $id, $settings) {
                $args = array($id);
                $self->invoke($name, $args, $settings)
                     ->then($topic->handler, $cb);
            };
            $topic->handler = function($result) use ($self, $name, $id, $cb) {
                $topic = $self->getTopic($name, $id);
                if ($topic !== null) {
                    if ($result !== null) {
                        $callbacks = $topic->callbacks;
                        foreach ($callbacks as $callback) {
                            try {
                                call_user_func($callback, $result);
                            }
                            catch (Exception $ex) {}
                            catch (Throwable $ex) {}
                        }
                    }
                    if ($self->getTopic($name, $id) !== null) $cb();
                }
            };
            $topic->callbacks = array($callback);
            $this->topics[$name][$id] = $topic;
            $cb();
        }
        elseif (array_search($callback, $topic->callbacks, true) === false) {
            $topic->callbacks[] = $callback;
        }
    }
345801
public function queueSubscribe($subject, $queue, \Closure $callback)
    {
        $sid = $this->randomGenerator->generateString(16);
        $msg = 'SUB '.$subject.' '.$queue.' '.$sid;
        $this->send($msg);
        $this->subscriptions[$sid] = $callback;
        return $sid;
    }
112580
private function handleSubscriptionUid(ConnectionInterface $from, $data) {
    if ($data['action'] === 'subscribe') {
      // Client is subscribing to a UID.
      if (!key_exists($data['uid'], $this->uidSubs)) {
        $this->uidSubs[$data['uid']] = new \SplObjectStorage();
      }
      $this->uidSubs[$data['uid']][$from] = [
        'count' => !!$data['count']
      ];
      $this->logger->notice(
          "Client subscribed to a UID! ".
            "({$data['uid']}, {$from->resourceId})"
      );

      if (Server::$config['broadcast_counts']) {
        // Notify clients of the subscription count.
        $count = count($this->uidSubs[$data['uid']]);
        foreach ($this->uidSubs[$data['uid']] as $key) {
          $curData = $this->uidSubs[$data['uid']][$key];
          if ($curData['count']) {
            $key->send(json_encode([
              'uid' => $data['uid'],
              'count' => $count
            ]));
          }
        }
      }
    }

    if ($data['action'] === 'unsubscribe') {
      // Client is unsubscribing from a UID.
      if (!key_exists($data['uid'], $this->uidSubs)) {
        return;
      }
      if (!$this->uidSubs[$data['uid']]->contains($from)) {
        return;
      }
      unset($this->uidSubs[$data['uid']][$from]);
      $this->logger->notice(
          "Client unsubscribed from a UID! ".
            "({$data['uid']}, {$from->resourceId})"
      );

      $count = count($this->uidSubs[$data['uid']]);

      if ($count === 0) {
        // No more subscribed clients.
        unset($this->uidSubs[$data['uid']]);
        return;
      }

      if (Server::$config['broadcast_counts']) {
        // Notify clients of the subscription count.
        $count = count($this->uidSubs[$data['uid']]);
        foreach ($this->uidSubs[$data['uid']] as $key) {
          $curData = $this->uidSubs[$data['uid']][$key];
          if ($curData['count']) {
            $key->send(json_encode([
              'uid' => $data['uid'],
              'count' => $count
            ]));
          }
        }
      }
    }
  }
1215997
function subscribe(subscriber, immediate) {
        if (!~subscribers.indexOf(subscriber)) {
            subscribers.push(subscriber);
        }
        if (immediate) {
            subscriber(value);
        }
    }
98825
public function addXPathHandler($xpath, $cb, $obj = null)
    {
        if (preg_match_all("/\(?{[^\}]+}\)?(\/?)[^\/]+/", $xpath, $regs)) {
            $ns_tags = $regs[0];
        } else {
            $ns_tags = [$xpath];
        }
        foreach ($ns_tags as $ns_tag) {
            list($l, $r) = explode("}", $ns_tag);
            if ($r !== null) {
                $xpart = [substr($l, 1), $r];
            } else {
                $xpart = [null, $l];
            }
            $xpath_array[] = $xpart;
        }
        $this->xpathhandlers[] = [$xpath_array, $cb, $obj, $xpath];
    }
548126
public function subscribe($subscriber)
    {
        if (is_object($subscriber)) {
            $subscriber->subscribe($this);
            return;
        }
        $this->subscribe(new $subscriber);
    }
443409
public function update(Listener $listener, $id, array $input)
    {
        $input['slug'] = $this->generateUniqueSlug($input);
        $validation    = $this->validator->on('update')->bind(['id' => $id])->with($input);

        if ($validation->fails()) {
            return $listener->updateHasFailedValidation($id, $validation->getMessageBag());
        }

        $content = Eloquent::findOrFail($id);

        $listener->authorize('update', $content);

        $this->saving($content, $input);

        return $listener->updateHasSucceed($content, $input);
    }