src/Performer.php
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