One Hat Cyber Team

Your IP: 216.73.216.45
Server IP: 185.238.29.86
Server: Linux server2 6.8.12-6-pve #1 SMP PREEMPT_DYNAMIC PMX 6.8.12-6 (2024-12-19T19:05Z) x86_64
Server Software: nginx/1.18.0
PHP Version: 8.1.31
Buat File | Buat Folder

Dir : ~/var/www/shopware/vendor/enqueue/amqp-tools/

View File Name : RabbitMqDelayPluginDelayStrategy.php

<?php

declare(strict_types=1);

namespace Enqueue\AmqpTools;

use Interop\Amqp\AmqpContext;
use Interop\Amqp\AmqpDestination;
use Interop\Amqp\AmqpMessage;
use Interop\Amqp\AmqpQueue;
use Interop\Amqp\AmqpTopic;
use Interop\Amqp\Impl\AmqpBind;
use Interop\Queue\Exception\InvalidDestinationException;

class RabbitMqDelayPluginDelayStrategy implements DelayStrategy
{
    public function delayMessage(AmqpContext $context, AmqpDestination $dest, AmqpMessage $message, int $delay): void
    {
        $delayMessage = $context->createMessage($message->getBody(), $message->getProperties(), $message->getHeaders());
        $delayMessage->setProperty('x-delay', (int) $delay);
        $delayMessage->setRoutingKey($message->getRoutingKey());

        if ($dest instanceof AmqpTopic) {
            $delayTopic = $context->createTopic('enqueue.'.$dest->getTopicName().'.delayed');
            $delayTopic->setType('x-delayed-message');
            $delayTopic->addFlag($dest->getFlags());
            $delayTopic->setArgument('x-delayed-type', $dest->getType());

            $context->declareTopic($delayTopic);
            $context->bind(new AmqpBind($dest, $delayTopic, $delayMessage->getRoutingKey()));
        } elseif ($dest instanceof AmqpQueue) {
            $delayTopic = $context->createTopic('enqueue.queue.delayed');
            $delayTopic->setType('x-delayed-message');
            $delayTopic->addFlag(AmqpTopic::FLAG_DURABLE);
            $delayTopic->setArgument('x-delayed-type', AmqpTopic::TYPE_DIRECT);

            $delayMessage->setRoutingKey($dest->getQueueName());

            $context->declareTopic($delayTopic);
            $context->bind(new AmqpBind($dest, $delayTopic, $delayMessage->getRoutingKey()));
        } else {
            throw new InvalidDestinationException(sprintf('The destination must be an instance of %s but got %s.',
                AmqpTopic::class.'|'.AmqpQueue::class,
                get_class($dest)
            ));
        }

        $producer = $context->createProducer();

        if ($producer instanceof DelayStrategyAware) {
            $producer->setDelayStrategy(null);
        }

        $producer->send($delayTopic, $delayMessage);
    }
}