public function waitAndExecute(string $queueName, $timeout = null): ?Message
    {

        $messageCacheIdentifier = null;
        $queue = $this->queueManager->getQueue($queueName);
        $message = $queue->waitAndReserve($timeout);
        if ($message === null) {
            $this->emitMessageTimeout($queue);
            // timeout
            return null;
        }
        $this->emitMessageReserved($queue, $message);

        $queueSettings = $this->queueManager->getQueueSettings($queueName);
        try {
            if (isset($queueSettings['executeIsolated']) && $queueSettings['executeIsolated'] === true) {
                $messageCacheIdentifier = sha1(serialize($message));
                $this->messageCache->set($messageCacheIdentifier, $message);
                Scripts::executeCommand('flowpack.jobqueue.common:job:execute', $this->flowSettings, false, ['queue' => $queue->getName(), 'messageCacheIdentifier' => $messageCacheIdentifier]);
            } else {
                $this->executeJobForMessage($queue, $message);
            }
        } catch (\Exception $exception) {
            $maximumNumberOfReleases = isset($queueSettings['maximumNumberOfReleases']) ? (integer)$queueSettings['maximumNumberOfReleases'] : 0;
            if ($message->getNumberOfReleases() < $maximumNumberOfReleases) {
                $releaseOptions = isset($queueSettings['releaseOptions']) ? $queueSettings['releaseOptions'] : [];
                $queue->release($message->getIdentifier(), $releaseOptions);
                $this->emitMessageReleased($queue, $message, $releaseOptions, $exception);
                throw new JobQueueException(sprintf('Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - RELEASE', $message->getIdentifier(), $queue->getName(), $message->getNumberOfReleases() + 1, $maximumNumberOfReleases + 1), 1334056583, $exception);
            } else {
                $queue->abort($message->getIdentifier());
                $this->emitMessageFailed($queue, $message, $exception);
                throw new JobQueueException(sprintf('Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - ABORTING', $message->getIdentifier(), $queue->getName(), $message->getNumberOfReleases() + 1, $maximumNumberOfReleases + 1), 1334056584, $exception);
            }
        } finally {
            if ($messageCacheIdentifier !== null) {
                $this->messageCache->remove($messageCacheIdentifier);
            }
        }

        $queue->finish($message->getIdentifier());
        $this->emitMessageFinished($queue, $message);

        return $message;
    }
public function waitAndExecute(string $queueName, $timeout = null): ?Message
    {

        $messageCacheIdentifier = null;
        $queue = $this->queueManager->getQueue($queueName);
        $message = $queue->waitAndReserve($timeout);
        if ($message === null) {
            $this->emitMessageTimeout($queue);
            // timeout
            return null;
        }
        $this->emitMessageReserved($queue, $message);

        $queueSettings = $this->queueManager->getQueueSettings($queueName);
        try {
            if (isset($queueSettings['executeIsolated']) && $queueSettings['executeIsolated'] === true) {
                $messageCacheIdentifier = sha1(serialize($message));
                $this->messageCache->set($messageCacheIdentifier, $message);
                Scripts::executeCommand('flowpack.jobqueue.common:job:execute', $this->flowSettings, false, ['queue' => $queue->getName(), 'messageCacheIdentifier' => $messageCacheIdentifier]);
            } else {
                $this->executeJobForMessage($queue, $message);
            }
        } catch (\Exception $exception) {
            $maximumNumberOfReleases = isset($queueSettings['maximumNumberOfReleases']) ? (integer)$queueSettings['maximumNumberOfReleases'] : 0;
            if ($message->getNumberOfReleases() < $maximumNumberOfReleases) {
                $releaseOptions = isset($queueSettings['releaseOptions']) ? $queueSettings['releaseOptions'] : [];
                $queue->release($message->getIdentifier(), $releaseOptions);
                $this->emitMessageReleased($queue, $message, $releaseOptions, $exception);
                throw new JobQueueException(sprintf('Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - RELEASE', $message->getIdentifier(), $queue->getName(), $message->getNumberOfReleases() + 1, $maximumNumberOfReleases + 1), 1334056583, $exception);
            } else {
                $queue->abort($message->getIdentifier());
                $this->emitMessageFailed($queue, $message, $exception);
                throw new JobQueueException(sprintf('Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - ABORTING', $message->getIdentifier(), $queue->getName(), $message->getNumberOfReleases() + 1, $maximumNumberOfReleases + 1), 1334056584, $exception);
            }
        } finally {
            if ($messageCacheIdentifier !== null) {
                $this->messageCache->remove($messageCacheIdentifier);
            }
        }

        $queue->finish($message->getIdentifier());
        $this->emitMessageFinished($queue, $message);

        return $message;
    }
public function waitAndExecute ( string $ queueName , $ timeout = null ) : ? Message { $ messageCacheIdentifier = null ; $ queue = $ this -> queueManager -> getQueue ( $ queueName ) ; $ message = $ queue -> waitAndReserve ( $ timeout ) ; if ( $ message === null ) { $ this -> emitMessageTimeout ( $ queue ) ; // timeout return null ; } $ this -> emitMessageReserved ( $ queue , $ message ) ; $ queueSettings = $ this -> queueManager -> getQueueSettings ( $ queueName ) ; try { if ( isset ( $ queueSettings [ 'executeIsolated' ] ) && $ queueSettings [ 'executeIsolated' ] === true ) { $ messageCacheIdentifier = sha1 ( serialize ( $ message ) ) ; $ this -> messageCache -> set ( $ messageCacheIdentifier , $ message ) ; Scripts :: executeCommand ( 'flowpack.jobqueue.common:job:execute' , $ this -> flowSettings , false , [ 'queue' => $ queue -> getName ( ) , 'messageCacheIdentifier' => $ messageCacheIdentifier ] ) ; } else { $ this -> executeJobForMessage ( $ queue , $ message ) ; } } catch ( \ Exception $ exception ) { $ maximumNumberOfReleases = isset ( $ queueSettings [ 'maximumNumberOfReleases' ] ) ? ( integer ) $ queueSettings [ 'maximumNumberOfReleases' ] : 0 ; if ( $ message -> getNumberOfReleases ( ) < $ maximumNumberOfReleases ) { $ releaseOptions = isset ( $ queueSettings [ 'releaseOptions' ] ) ? $ queueSettings [ 'releaseOptions' ] : [ ] ; $ queue -> release ( $ message -> getIdentifier ( ) , $ releaseOptions ) ; $ this -> emitMessageReleased ( $ queue , $ message , $ releaseOptions , $ exception ) ; throw new JobQueueException ( sprintf ( 'Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - RELEASE' , $ message -> getIdentifier ( ) , $ queue -> getName ( ) , $ message -> getNumberOfReleases ( ) + 1 , $ maximumNumberOfReleases + 1 ) , 1334056583 , $ exception ) ; } else { $ queue -> abort ( $ message -> getIdentifier ( ) ) ; $ this -> emitMessageFailed ( $ queue , $ message , $ exception ) ; throw new JobQueueException ( sprintf ( 'Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - ABORTING' , $ message -> getIdentifier ( ) , $ queue -> getName ( ) , $ message -> getNumberOfReleases ( ) + 1 , $ maximumNumberOfReleases + 1 ) , 1334056584 , $ exception ) ; } } finally { if ( $ messageCacheIdentifier !== null ) { $ this -> messageCache -> remove ( $ messageCacheIdentifier ) ; } } $ queue -> finish ( $ message -> getIdentifier ( ) ) ; $ this -> emitMessageFinished ( $ queue , $ message ) ; return $ message ; }
Wait for a job in the given queue and execute it
A worker using this method should catch exceptions

@param string $queueName
@param integer $timeout
@return Message The message that was processed or NULL if no job was executed and a timeout occurred
@throws \Exception
@api
Wait for a job in the given queue and execute it A worker using this method should catch exceptions