public function waitAllowingSubscriptionMessage()
    {
        if (!$this->subscriberSocket) {
            $this->subscriberSocket = $this->context->getSocket(\ZMQ::SOCKET_SUB);
            $this->subscriberSocket->connect($this->getSocketsParams()->getPublisherPulsarSocketAddress());
        }

        $this->subscriberSocket->setSockOpt(\ZMQ::SOCKOPT_SUBSCRIBE, "");
        $this->logger->debug("Performer become subscriber.");

        //use in Pulsar to divide termination of current subscribers and potential subscribers, if early termination occurred
        $this->performerEarlyTerminated->setStandOnSubscription(true);

        $this->pushReadyToGetSubscriptionMsg();

        //->recv() block execution until data will received; all subscribers will continue execution synchronously
        $this->logger->debug("Performer wait subscription msg.");
        $this->publisherToSubscribersDto = unserialize($this->subscriberSocket->recv());
        $this->logger->debug("Performer got subscription msg.");

        if (!$this->publisherToSubscribersDto->isAllowAction()) {
            throw new PublisherPulsarException(PublisherPulsarExceptionsConstants::ACTION_IS_NOT_ALLOWED);
        }

        return null;
    }
public function waitAllowingSubscriptionMessage()
    {
        if (!$this->subscriberSocket) {
            $this->subscriberSocket = $this->context->getSocket(\ZMQ::SOCKET_SUB);
            $this->subscriberSocket->connect($this->getSocketsParams()->getPublisherPulsarSocketAddress());
        }

        $this->subscriberSocket->setSockOpt(\ZMQ::SOCKOPT_SUBSCRIBE, "");
        $this->logger->debug("Performer become subscriber.");

        //use in Pulsar to divide termination of current subscribers and potential subscribers, if early termination occurred
        $this->performerEarlyTerminated->setStandOnSubscription(true);

        $this->pushReadyToGetSubscriptionMsg();

        //->recv() block execution until data will received; all subscribers will continue execution synchronously
        $this->logger->debug("Performer wait subscription msg.");
        $this->publisherToSubscribersDto = unserialize($this->subscriberSocket->recv());
        $this->logger->debug("Performer got subscription msg.");

        if (!$this->publisherToSubscribersDto->isAllowAction()) {
            throw new PublisherPulsarException(PublisherPulsarExceptionsConstants::ACTION_IS_NOT_ALLOWED);
        }

        return null;
    }
public function waitAllowingSubscriptionMessage ( ) { if ( ! $ this -> subscriberSocket ) { $ this -> subscriberSocket = $ this -> context -> getSocket ( \ ZMQ :: SOCKET_SUB ) ; $ this -> subscriberSocket -> connect ( $ this -> getSocketsParams ( ) -> getPublisherPulsarSocketAddress ( ) ) ; } $ this -> subscriberSocket -> setSockOpt ( \ ZMQ :: SOCKOPT_SUBSCRIBE , "" ) ; $ this -> logger -> debug ( "Performer become subscriber." ) ; //use in Pulsar to divide termination of current subscribers and potential subscribers, if early termination occurred $ this -> performerEarlyTerminated -> setStandOnSubscription ( true ) ; $ this -> pushReadyToGetSubscriptionMsg ( ) ; //->recv() block execution until data will received; all subscribers will continue execution synchronously $ this -> logger -> debug ( "Performer wait subscription msg." ) ; $ this -> publisherToSubscribersDto = unserialize ( $ this -> subscriberSocket -> recv ( ) ) ; $ this -> logger -> debug ( "Performer got subscription msg." ) ; if ( ! $ this -> publisherToSubscribersDto -> isAllowAction ( ) ) { throw new PublisherPulsarException ( PublisherPulsarExceptionsConstants :: ACTION_IS_NOT_ALLOWED ) ; } return null ; }
Wait subscription message before doing action
@return null
Wait subscription message before doing action