Components/Queues/Drivers/AmqpQueueDriver.php
public function receiveNoWait()
    {
        if ($this->currentEnvelope) {
            throw new \RuntimeException(
                'AmqpQueueDriver: This driver has a message that was not acknowledged yet. A message must be processed and acknowledged before receiving new messages.'
            );
        }

        $this->dequeueingTimeMs = 0;

        $this->currentEnvelope = $this->queue->get();

        $msg = null;

        if (!$this->currentEnvelope) {
            return null;
        }

        $start = microtime(true);
        $deserializationContext = new DeserializationContext();

        /** @var QueueMessageInterface $msg */
        $msg = $this->getSerializer()->deserialize($this->currentEnvelope->getBody(), SerializableInterface::class, $this->format, $deserializationContext);

        foreach ($this->currentEnvelope->getHeaders() as $header => $value) {
            $msg->setHeader($header, $value);
        }

        // Calculate how long it took to deserilize the message
        $this->dequeueingTimeMs = (int) ((microtime(true) - $start) * 1000);

        return $msg;
    }
public function receiveNoWait()
    {
        if ($this->currentEnvelope) {
            throw new \RuntimeException(
                'AmqpQueueDriver: This driver has a message that was not acknowledged yet. A message must be processed and acknowledged before receiving new messages.'
            );
        }

        $this->dequeueingTimeMs = 0;

        $this->currentEnvelope = $this->queue->get();

        $msg = null;

        if (!$this->currentEnvelope) {
            return null;
        }

        $start = microtime(true);
        $deserializationContext = new DeserializationContext();

        /** @var QueueMessageInterface $msg */
        $msg = $this->getSerializer()->deserialize($this->currentEnvelope->getBody(), SerializableInterface::class, $this->format, $deserializationContext);

        foreach ($this->currentEnvelope->getHeaders() as $header => $value) {
            $msg->setHeader($header, $value);
        }

        // Calculate how long it took to deserilize the message
        $this->dequeueingTimeMs = (int) ((microtime(true) - $start) * 1000);

        return $msg;
    }
public function receiveNoWait ( ) { if ( $ this -> currentEnvelope ) { throw new \ RuntimeException ( 'AmqpQueueDriver: This driver has a message that was not acknowledged yet. A message must be processed and acknowledged before receiving new messages.' ) ; } $ this -> dequeueingTimeMs = 0 ; $ this -> currentEnvelope = $ this -> queue -> get ( ) ; $ msg = null ; if ( ! $ this -> currentEnvelope ) { return null ; } $ start = microtime ( true ) ; $ deserializationContext = new DeserializationContext ( ) ; /** @var QueueMessageInterface $msg */ $ msg = $ this -> getSerializer ( ) -> deserialize ( $ this -> currentEnvelope -> getBody ( ) , SerializableInterface :: class , $ this -> format , $ deserializationContext ) ; foreach ( $ this -> currentEnvelope -> getHeaders ( ) as $ header => $ value ) { $ msg -> setHeader ( $ header , $ value ) ; } // Calculate how long it took to deserilize the message $ this -> dequeueingTimeMs = ( int ) ( ( microtime ( true ) - $ start ) * 1000 ) ; return $ msg ; }
Returns One Serializable object from the queue.

It requires to subscribe previously to a specific queue

@throws \Exception

@return \Smartbox\Integration\FrameworkBundle\Components\Queues\QueueMessageInterface|null
Returns One Serializable object from the queue .