src/console/QueueController.php
private function requeue(string $queueName, AMQPMessage $msg, NotProcessableException $exception): void
    {
        $tries = 0;
        $headers = $msg->get_properties()['application_headers'];
        if ($headers instanceof AMQPTable) {
            $tries = $headers->getNativeData()['x-number-of-tries'] ?? 0;
        }

        if ($exception->getMaxTries() !== null && $tries >= $exception->getMaxTries()) {
            $this->logger->debug('No tries left for message. Marking it as an error', ['message' => $msg, 'exception' => $exception]);
            $this->handleError($queueName, $msg, $exception);
            return;
        }

        // Init delay exchange
        $channel = $this->amqp->channel();
        $delayExchange = "$queueName.delayed";
        $channel->exchange_declare($delayExchange, 'x-delayed-message', false, true, true, false, false, new AMQPTable([
            'x-delayed-type' => 'direct',
        ]));
        $channel->queue_bind($queueName, $delayExchange);

        // Send message
        $delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)pow($exception->getProgressionMultiplier(), $tries);
        $delayMessage = new AMQPMessage($msg->getBody(), array_merge($msg->get_properties(), [
            'application_headers' => new AMQPTable([
                'x-delay' => $delayDuration,
                'x-number-of-tries' => $tries + 1,
            ]),
        ]));
        $channel->basic_publish($delayMessage, $delayExchange, '');
        $this->logger->debug('Delayed message for ' . $delayDuration, ['message' => $msg, 'exception' => $exception]);
    }
private function requeue(string $queueName, AMQPMessage $msg, NotProcessableException $exception): void
    {
        $tries = 0;
        $headers = $msg->get_properties()['application_headers'];
        if ($headers instanceof AMQPTable) {
            $tries = $headers->getNativeData()['x-number-of-tries'] ?? 0;
        }

        if ($exception->getMaxTries() !== null && $tries >= $exception->getMaxTries()) {
            $this->logger->debug('No tries left for message. Marking it as an error', ['message' => $msg, 'exception' => $exception]);
            $this->handleError($queueName, $msg, $exception);
            return;
        }

        // Init delay exchange
        $channel = $this->amqp->channel();
        $delayExchange = "$queueName.delayed";
        $channel->exchange_declare($delayExchange, 'x-delayed-message', false, true, true, false, false, new AMQPTable([
            'x-delayed-type' => 'direct',
        ]));
        $channel->queue_bind($queueName, $delayExchange);

        // Send message
        $delayDuration = 1000 * $exception->getSecondsBeforeRetry() * (int)pow($exception->getProgressionMultiplier(), $tries);
        $delayMessage = new AMQPMessage($msg->getBody(), array_merge($msg->get_properties(), [
            'application_headers' => new AMQPTable([
                'x-delay' => $delayDuration,
                'x-number-of-tries' => $tries + 1,
            ]),
        ]));
        $channel->basic_publish($delayMessage, $delayExchange, '');
        $this->logger->debug('Delayed message for ' . $delayDuration, ['message' => $msg, 'exception' => $exception]);
    }
private function requeue ( string $ queueName , AMQPMessage $ msg , NotProcessableException $ exception ) : void { $ tries = 0 ; $ headers = $ msg -> get_properties ( ) [ 'application_headers' ] ; if ( $ headers instanceof AMQPTable ) { $ tries = $ headers -> getNativeData ( ) [ 'x-number-of-tries' ] ?? 0 ; } if ( $ exception -> getMaxTries ( ) !== null && $ tries >= $ exception -> getMaxTries ( ) ) { $ this -> logger -> debug ( 'No tries left for message. Marking it as an error' , [ 'message' => $ msg , 'exception' => $ exception ] ) ; $ this -> handleError ( $ queueName , $ msg , $ exception ) ; return ; } // Init delay exchange $ channel = $ this -> amqp -> channel ( ) ; $ delayExchange = "$queueName.delayed" ; $ channel -> exchange_declare ( $ delayExchange , 'x-delayed-message' , false , true , true , false , false , new AMQPTable ( [ 'x-delayed-type' => 'direct' , ] ) ) ; $ channel -> queue_bind ( $ queueName , $ delayExchange ) ; // Send message $ delayDuration = 1000 * $ exception -> getSecondsBeforeRetry ( ) * ( int ) pow ( $ exception -> getProgressionMultiplier ( ) , $ tries ) ; $ delayMessage = new AMQPMessage ( $ msg -> getBody ( ) , array_merge ( $ msg -> get_properties ( ) , [ 'application_headers' => new AMQPTable ( [ 'x-delay' => $ delayDuration , 'x-number-of-tries' => $ tries + 1 , ] ) , ] ) ) ; $ channel -> basic_publish ( $ delayMessage , $ delayExchange , '' ) ; $ this -> logger -> debug ( 'Delayed message for ' . $ delayDuration , [ 'message' => $ msg , 'exception' => $ exception ] ) ; }
Resends message to queue with a delay

@param string $queueName
@param AMQPMessage $msg
@param NotProcessableException $exception
Resends message to queue with a delay