Subscribes to a topic filter.

@param Subscription $subscription

@return ExtendedPromiseInterface
182115
public function subscribe(Subscription $subscription)
    {
        if (!$this->isConnected) {
            return new RejectedPromise(new \LogicException('The client is not connected.'));
        }

        return $this->startFlow(new OutgoingSubscribeFlow([$subscription], $this->identifierGenerator));
    }
590746
public function subscriptions(Subscriber $subscriber): Collection
    {
        // A subscription can be fired w/out a request so we must make
        // sure the schema has been generated.
        $this->graphQL->prepSchema();

        return (new Collection($subscriber->query->definitions))
            ->filter(function (Node $node): bool {
                return $node instanceof OperationDefinitionNode;
            })
            ->filter(function (OperationDefinitionNode $node): bool {
                return $node->operation === 'subscription';
            })
            ->flatMap(function (OperationDefinitionNode $node) {
                return (new Collection($node->selectionSet->selections))
                    ->map(function (FieldNode $field): string {
                        return $field->name->value;
                    })
                    ->toArray();
            })
            ->map(function ($subscriptionField): GraphQLSubscription {
                return Arr::get(
                    $this->subscriptions,
                    $subscriptionField,
                    new NotFoundSubscription
                );
            });
    }
153114
private function changeSubscription($subscription, $type)
    {
        if ($subscription->getSubscriberEmail()) {
            $this
                ->filterEmail($subscription->getSubscriberEmail())
                ->filterCustomer($subscription->getCustomerId());

            if ($this->customerModelConfigShare->isWebsiteScope()) {
                $this
                    ->joinWebsite()
                    ->filterWebsite($subscription->getWebsiteId());
            }

            /** @var Subscriber $subscriber |bool */
            $subscriber = $this->subscriptionCollection->fetchItem();

            if (!$subscriber) {
                if ($type !== Subscriber::STATUS_SUBSCRIBED) {
                    return false;
                }

                $subscriber = $this->subscriberFactory->create();
            }

            foreach ($subscription->getData() as $key => $value) {
                $subscriber->setData($key, $value);
            }

            $subscriber->setStatus($type);
            $subscriber->setStatusChanged(true);

            $subscriber->save();
            return true;
        }
        return false;
    }
1278353
public function subscribe(string $topicName, string $subscribeName, $endPoint, $strategy = null, $contentFormat = null, $topicOwner = null, $createTime = null, $lastModifyTime = null) {
        if(!$topicName || !$subscribeName || !$endPoint){
            return $this->error('参数有误');
        }
        $topic = $this->client->getTopicRef($topicName);
        $attributes = new SubscriptionAttributes($subscribeName, $endPoint, $strategy, $contentFormat, $topicName, $topicOwner, $createTime, $lastModifyTime);
        try{
            $topic->subscribe($attributes);
            return $this->success('创建订阅成功');
        }catch(MnsException $exception){
            return $this->parseException($exception, '创建订阅');
        }
    }
637301
public function triggerSubscriptionRebill(GatewayInterface $gateway, SubscriptionInterface $subscription, RebillInterface $rebill)
    {
        $this->trigger(self::ON_SUBSCRIPTION_REBILLED, $gateway, $subscription, $rebill);
    }
728919
public function getSubscription($topicPath, $subscriptionName)
    {
        $httpCallContext = new HttpCallContext();
        $httpCallContext->setMethod(Resources::HTTP_GET);
        $httpCallContext->addStatusCode(Resources::STATUS_OK);
        $subscriptionPath = sprintf(
            Resources::SUBSCRIPTION_PATH,
            $topicPath,
            $subscriptionName
        );
        $httpCallContext->setPath($subscriptionPath);
        $response = $this->sendHttpContext($httpCallContext);
        $subscriptionInfo = new SubscriptionInfo();
        $subscriptionInfo->parseXml($response->getBody());

        return $subscriptionInfo;
    }
278818
public function getSubscription($subscriptionId, array $parameters = [])
    {
        return $this->client->subscriptions->getFor($this, $subscriptionId, $parameters);
    }
1093258
public function addEventSubscriber(AnEventSubscriberInterface $subscriber, $priority = null)
    {
        $handle = $subscriber->getHandle();
    
        if (! isset($this->_subscribers[$handle])) {
            $subscriptions = $subscriber->getSubscriptions();
            $priority = is_int($priority) ? $priority : $subscriber->getPriority();
    
            foreach ($subscriptions as $subscription) {
                $this->addEventListener($subscription, $subscriber, $priority);
            }
    
            $this->_subscribers[$handle] = $subscriber;
        }
    
        return $this;
    }
712888
public function subscriptions()
    {
        return $this->client->subscriptions->listFor($this, null, null, $this->getPresetOptions());
    }
1297662
public function register(GraphQLSubscription $subscription, string $field): self
    {
        $this->subscriptions[$field] = $subscription;

        return $this;
    }
1095884
public function subscribe(string $eventName, callable $subscriber): self
    {
        if ($subscriber instanceof Closure || ($staticMethod = $this->parseStaticMethod($subscriber)) === null) {
            $this->subscribers[$eventName][] = $subscriber;
        } else {
            assert($staticMethod !== null);
            $this->subscribers[$eventName][] = $this->getUnifiedStaticMethodRepresentation($staticMethod);
        }

        $this->addAllowedEvent($eventName);

        return $this;
    }
1309650
public function has_subscriber($subscriptions)
    {
        $enabled_components = $this->wrappedObject->enabled_components()->orderBy('order')->pluck('id')->toArray();
        $intersected = array_intersect($enabled_components, $subscriptions);

        return count($intersected) != 0;
    }
979467
public static function toObservable(PromiseInterface $promise): Observable
    {
        $subject = new AsyncSubject();

        $p = $promise->then(
            function ($value) use ($subject) {
                $subject->onNext($value);
                $subject->onCompleted();
            },
            function ($error) use ($subject) {
                $error = $error instanceof \Exception ? $error : new RejectedPromiseException($error);
                $subject->onError($error);
            }
        );

        return new AnonymousObservable(function ($observer) use ($subject, $p) {
            $disp = $subject->subscribe($observer);
            return new CallbackDisposable(function () use ($p, $disp) {
                $disp->dispose();
                if ($p instanceof CancellablePromiseInterface) {
                    $p->cancel();
                }
            });
        });
    }
265712
public function publishState(Subscription $subscription)
    {
        //Pause all non-state building event messages
        $subscription->pauseForState();

        $sessionId = $subscription->getSession()->getSessionId();

        $this->clientSession->call($this->getProcedureName(),
            [$subscription->getUri(), $sessionId, $subscription->getOptions(), $subscription->getSession()->getAuthenticationDetails()])->then(
            function ($res) use ($subscription) {
                $pubId = null;
                if (isset($res[0])) {
                    $pubId = $res[0];
                }
                $subscription->unPauseForState($pubId);
            },
            function ($error) use ($subscription) {
                Logger::error($this, "Could not call '{$this->getProcedureName()}'");
                $subscription->unPauseForState();
            }
        );

    }
612021
public function isSubscribed(AnEventSubscriberInterface $subscriber)
    {
        $handle = $subscriber->getHandle();
        return isset($this->_subscribers[$handle]);
    }
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;
    }
1189594
public function createSubscription(array $options = [], array $filters = [])
    {
        return $this->client->subscriptions->createFor($this, $options, $filters);
    }