library/AbstractWorkerController.php
public function await(QueueClientInterface $queue, ReceiveParametersInterface $params = null)
    {
        $worker = $this;
        $this->await = true;
        $lastResult = [];

        $callback = function(QueueEvent $event) use($worker, $queue, &$lastResult) {

            $messages = $event->getMessages();

            foreach ($messages as $message) {
                $lastResult = $worker->process($message);

                if ($queue->canDeleteMessage()) {
                    $queue->delete($message);
                }
            }

            if ($worker->isAwaitingStopped()) {
                $event->stopAwait(true);
            }

            return !$worker->isAwaitingStopped();
        };


        $callbackHandler = $queue->getEventManager()->attach(QueueEvent::EVENT_RECEIVE, $callback);

        $queue->await($params);

        $queue->getEventManager()->detach($callbackHandler);


        return $lastResult;
    }
public function await(QueueClientInterface $queue, ReceiveParametersInterface $params = null)
    {
        $worker = $this;
        $this->await = true;
        $lastResult = [];

        $callback = function(QueueEvent $event) use($worker, $queue, &$lastResult) {

            $messages = $event->getMessages();

            foreach ($messages as $message) {
                $lastResult = $worker->process($message);

                if ($queue->canDeleteMessage()) {
                    $queue->delete($message);
                }
            }

            if ($worker->isAwaitingStopped()) {
                $event->stopAwait(true);
            }

            return !$worker->isAwaitingStopped();
        };


        $callbackHandler = $queue->getEventManager()->attach(QueueEvent::EVENT_RECEIVE, $callback);

        $queue->await($params);

        $queue->getEventManager()->detach($callbackHandler);


        return $lastResult;
    }
public function await ( QueueClientInterface $ queue , ReceiveParametersInterface $ params = null ) { $ worker = $ this ; $ this -> await = true ; $ lastResult = [ ] ; $ callback = function ( QueueEvent $ event ) use ( $ worker , $ queue , & $ lastResult ) { $ messages = $ event -> getMessages ( ) ; foreach ( $ messages as $ message ) { $ lastResult = $ worker -> process ( $ message ) ; if ( $ queue -> canDeleteMessage ( ) ) { $ queue -> delete ( $ message ) ; } } if ( $ worker -> isAwaitingStopped ( ) ) { $ event -> stopAwait ( true ) ; } return ! $ worker -> isAwaitingStopped ( ) ; } ; $ callbackHandler = $ queue -> getEventManager ( ) -> attach ( QueueEvent :: EVENT_RECEIVE , $ callback ) ; $ queue -> await ( $ params ) ; $ queue -> getEventManager ( ) -> detach ( $ callbackHandler ) ; return $ lastResult ; }
Wait for and process incoming messages

TODO: handle idle
@param QueueClientInterface $queue
@param ReceiveParametersInterface $params
@return boolean|\Stakhanovist\Worker\mixed
Wait for and process incoming messages