Core/Handlers/MessageHandler.php
public function handle(MessageInterface $message, EndpointInterface $endpointFrom)
    {
        $retries = 0;
        // If this is an exchange envelope, we extract the old exchange and prepare the new one
        if ($message && $message instanceof ExchangeEnvelope) {
            $oldExchange = $message->getBody();
            $exchange = new Exchange($oldExchange->getIn(), $oldExchange->getItinerary(), $oldExchange->getHeaders());

            if ($message instanceof RetryExchangeEnvelope) {
                $retries = $message->getRetries();
                $delaySinceLastRetry = round(microtime(true) * 1000) - $message->getHeader(RetryExchangeEnvelope::HEADER_LAST_RETRY_AT);
                $retryDelay = $message->getHeader(RetryExchangeEnvelope::HEADER_RETRY_DELAY) * 1000;

                $endpointURI = $message instanceof ThrottledExchangeEnvelope ? $this->throttleURI : $this->retryURI;

                if ($delaySinceLastRetry < $retryDelay) {
                    $this->deferExchangeMessage($message, $endpointURI);

                    return;
                }
            }
        }
        // Otherwise create the exchange
        else {
            $exchange = $this->createExchangeForMessageFromURI($message, $endpointFrom->getURI());
        }

        $this->onHandleStart($exchange);
        $result = $this->processExchange($exchange, $retries);
        $this->onHandleSuccess($exchange);
        return $result;
    }
public function handle(MessageInterface $message, EndpointInterface $endpointFrom)
    {
        $retries = 0;
        // If this is an exchange envelope, we extract the old exchange and prepare the new one
        if ($message && $message instanceof ExchangeEnvelope) {
            $oldExchange = $message->getBody();
            $exchange = new Exchange($oldExchange->getIn(), $oldExchange->getItinerary(), $oldExchange->getHeaders());

            if ($message instanceof RetryExchangeEnvelope) {
                $retries = $message->getRetries();
                $delaySinceLastRetry = round(microtime(true) * 1000) - $message->getHeader(RetryExchangeEnvelope::HEADER_LAST_RETRY_AT);
                $retryDelay = $message->getHeader(RetryExchangeEnvelope::HEADER_RETRY_DELAY) * 1000;

                $endpointURI = $message instanceof ThrottledExchangeEnvelope ? $this->throttleURI : $this->retryURI;

                if ($delaySinceLastRetry < $retryDelay) {
                    $this->deferExchangeMessage($message, $endpointURI);

                    return;
                }
            }
        }
        // Otherwise create the exchange
        else {
            $exchange = $this->createExchangeForMessageFromURI($message, $endpointFrom->getURI());
        }

        $this->onHandleStart($exchange);
        $result = $this->processExchange($exchange, $retries);
        $this->onHandleSuccess($exchange);
        return $result;
    }
public function handle ( MessageInterface $ message , EndpointInterface $ endpointFrom ) { $ retries = 0 ; // If this is an exchange envelope, we extract the old exchange and prepare the new one if ( $ message && $ message instanceof ExchangeEnvelope ) { $ oldExchange = $ message -> getBody ( ) ; $ exchange = new Exchange ( $ oldExchange -> getIn ( ) , $ oldExchange -> getItinerary ( ) , $ oldExchange -> getHeaders ( ) ) ; if ( $ message instanceof RetryExchangeEnvelope ) { $ retries = $ message -> getRetries ( ) ; $ delaySinceLastRetry = round ( microtime ( true ) * 1000 ) - $ message -> getHeader ( RetryExchangeEnvelope :: HEADER_LAST_RETRY_AT ) ; $ retryDelay = $ message -> getHeader ( RetryExchangeEnvelope :: HEADER_RETRY_DELAY ) * 1000 ; $ endpointURI = $ message instanceof ThrottledExchangeEnvelope ? $ this -> throttleURI : $ this -> retryURI ; if ( $ delaySinceLastRetry < $ retryDelay ) { $ this -> deferExchangeMessage ( $ message , $ endpointURI ) ; return ; } } } // Otherwise create the exchange else { $ exchange = $ this -> createExchangeForMessageFromURI ( $ message , $ endpointFrom -> getURI ( ) ) ; } $ this -> onHandleStart ( $ exchange ) ; $ result = $ this -> processExchange ( $ exchange , $ retries ) ; $ this -> onHandleSuccess ( $ exchange ) ; return $ result ; }
Handle a message.

If the message is a retryable message and the message is not ready to be processed yet, we will re-defer the message.
Otherwise process the message by putting it as part of an exchange, and processing the exchange.

{@inheritdoc}
Handle a message .