Yiisoft\Queue\AMQP\Middleware\DelayMiddleware

Complexity

Total Complexity

9

Size/Duplication

Total Lines	91
Duplicated Lines	0 %

Importance

Changes

0

<?php
final class DelayMiddleware implements DelayMiddlewareInterface
{
    public function __construct(private float $delayInSeconds, private bool $forcePersistentMessages = true)
    {
    }

    /**
     * @param float $seconds
     *
     * @return $this
     */
    public function withDelay(float $seconds): self
    {
        $new = clone $this;
        $new->delayInSeconds = $seconds;

        return $new;
    }

    public function getDelay(): float
    {
        return $this->delayInSeconds;
    }

    public function processPush(PushRequest $request, MessageHandlerPushInterface $handler): PushRequest
    {
        $adapter = $request->getAdapter();
        if (!$adapter instanceof Adapter) {
            throw new InvalidArgumentException(
                sprintf(
                    'This middleware works only with the %s. %s given.',
                    Adapter::class,
                    get_debug_type($adapter)
                )
            );
        }

        $queueProvider = $adapter->getQueueProvider();
        $originalExchangeSettings = $queueProvider->getExchangeSettings();
        $delayedExchangeSettings = $this->getExchangeSettings($originalExchangeSettings);
        $queueSettings = $this->getQueueSettings(
            $queueProvider->getQueueSettings(),
            $originalExchangeSettings
        );

        $adapter = $adapter->withQueueProvider(
            $queueProvider
                ->withMessageProperties($this->getMessageProperties($queueProvider))
                ->withExchangeSettings($delayedExchangeSettings)
                ->withQueueSettings($queueSettings)
        );

        return $handler->handlePush($request->withAdapter($adapter));
    }

    /**
     * @psalm-return array{expiration: int|float, delivery_mode?: int}&array
     */
    private function getMessageProperties(QueueProviderInterface $queueProvider): array
    {
        $messageProperties = ['expiration' => $this->delayInSeconds * 1000];
        if ($this->forcePersistentMessages === true) {
            $messageProperties['delivery_mode'] = AMQPMessage::DELIVERY_MODE_PERSISTENT;
        }

        return array_merge($queueProvider->getMessageProperties(), $messageProperties);
    }

    private function getQueueSettings(
        QueueSettingsInterface $queueSettings,
        ?ExchangeSettingsInterface $originalExchangeSettings
    ): QueueSettingsInterface {
        $arguments = [
            'x-dead-letter-exchange' => ['S', $originalExchangeSettings?->getName() ?? ''],
            'x-dead-routing-key' => ['S', $queueSettings->getName()],
            'x-expires' => ['I', $this->delayInSeconds * 1000 + 30_000],

A parse error occurred: Syntax error, unexpected T_STRING, expecting ',' or ']' on line 93 at column 66

            'x-message-ttl' => ['I', $this->delayInSeconds * 1000],
        ];
        return $queueSettings
            ->withName("{$queueSettings->getName()}.dlx")
            ->withArguments($arguments);
    }

    /**
     * @see https://github.com/vimeo/psalm/issues/9454
     *
     * @psalm-suppress LessSpecificReturnType
     */
    private function getExchangeSettings(?ExchangeSettingsInterface $exchangeSettings): ?ExchangeSettingsInterface
    {
        /** @noinspection NullPointerExceptionInspection */
        return $exchangeSettings