feat: symfony messenger delays

Author: dgafka

packages/Laravel/tests/Queue/LaravelQueueFinalFailureStrategyTest.php

@@ -4,14 +4,17 @@
 
 namespace Test\Ecotone\Laravel\Queue;
 
+use DateTimeImmutable;
 use Ecotone\Laravel\Queue\LaravelQueueMessageChannelBuilder;
 use Ecotone\Lite\EcotoneLite;
 use Ecotone\Messaging\Attribute\Asynchronous;
 use Ecotone\Messaging\Config\ConfigurationException;
 use Ecotone\Messaging\Config\ServiceConfiguration;
 use Ecotone\Messaging\Endpoint\ExecutionPollingMetadata;
 use Ecotone\Messaging\Endpoint\FinalFailureStrategy;
+use Ecotone\Messaging\MessageHeaders;
 use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Modelling\Attribute\QueryHandler;
 use Exception;
 use Illuminate\Database\Schema\Blueprint;
 use Illuminate\Foundation\Http\Kernel;
@@ -84,6 +87,29 @@ public function test_release_failure_strategy_releases_message_on_exception()
         );
     }
 
+    public function test_sending_with_delay_using_datetime()
+    {
+        $container = $this->getContainer();
+        $delayedService = new DelayedService();
+        $container->instance(DelayedService::class, $delayedService);
+
+        $ecotoneTestSupport = EcotoneLite::bootstrapFlowTesting(
+            [DelayedService::class],
+            $container,
+            ServiceConfiguration::createWithAsynchronicityOnly()
+                ->withExtensionObjects([
+                    LaravelQueueMessageChannelBuilder::create('async', 'database'),
+                ])
+        );
+
+        $ecotoneTestSupport->sendCommandWithRoutingKey('execute.delayed_command', new DelayedCommand('test_1'), metadata: [
+            MessageHeaders::DELIVERY_DELAY => (new DateTimeImmutable())->modify('+1 second'),
+        ]);
+        $ecotoneTestSupport->run('async', ExecutionPollingMetadata::createWithTestingSetup(maxExecutionTimeInMilliseconds: 2000));
+
+        $this->assertEquals(['test_1'], $delayedService->getMessages());
+    }
+
     private function getContainer(): ContainerInterface
     {
         $app = require __DIR__ . '/../Application/bootstrap/app.php';
@@ -109,3 +135,32 @@ public function execute(FailingCommand $command): void
         throw new Exception('Failing');
     }
 }
+
+class DelayedCommand
+{
+    public function __construct(public readonly string $payload)
+    {
+    }
+}
+
+class DelayedService
+{
+    /** @var string[] */
+    private array $messages = [];
+
+    #[Asynchronous('async')]
+    #[CommandHandler('execute.delayed_command', 'delayed_endpoint')]
+    public function execute(DelayedCommand $command): void
+    {
+        $this->messages[] = $command->payload;
+    }
+
+    /**
+     * @return string[]
+     */
+    #[QueryHandler('get.delayed_messages')]
+    public function getMessages(): array
+    {
+        return $this->messages;
+    }
+}

packages/Symfony/.gitignore

@@ -8,4 +8,8 @@ composer.lock
 phpunit.xml
 # Ignore generated container files
 /var/cache/
-/tests/phpunit/*/var/cache/
+/tests/phpunit/*/var/cache/
+# Ignore generated reference files
+config/reference.php
+tests/phpunit/config/reference.php
+tests/phpunit/*/config/reference.php

packages/Symfony/SymfonyBundle/Messenger/SymfonyMessageConverter.php

@@ -4,13 +4,17 @@
 
 namespace Ecotone\SymfonyBundle\Messenger;
 
+use DateTimeInterface;
 use Ecotone\Messaging\Conversion\ConversionService;
 use Ecotone\Messaging\Conversion\MediaType;
 use Ecotone\Messaging\Endpoint\FinalFailureStrategy;
 use Ecotone\Messaging\Handler\Type;
 use Ecotone\Messaging\Message;
 use Ecotone\Messaging\MessageConverter\HeaderMapper;
 use Ecotone\Messaging\MessageHeaders;
+use Ecotone\Messaging\Scheduling\DatePoint;
+use Ecotone\Messaging\Scheduling\Duration;
+use Ecotone\Messaging\Scheduling\TimeSpan;
 use Ecotone\Messaging\Support\MessageBuilder;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Stamp\DelayStamp;
@@ -53,12 +57,39 @@ public function convertToSymfonyMessage(Message $message, bool $withDelay): Enve
         $envelopeToSend = new Envelope($payload, [new MetadataStamp($headers)]);
 
         if ($message->getHeaders()->containsKey(MessageHeaders::DELIVERY_DELAY) && $withDelay) {
-            $envelopeToSend = $envelopeToSend->with(new DelayStamp($message->getHeaders()->get(MessageHeaders::DELIVERY_DELAY)));
+            $deliveryDelay = $this->convertDeliveryDelayToMilliseconds(
+                $message->getHeaders()->get(MessageHeaders::DELIVERY_DELAY),
+                $message->getHeaders()->getTimestamp()
+            );
+            if ($deliveryDelay !== null) {
+                $envelopeToSend = $envelopeToSend->with(new DelayStamp($deliveryDelay));
+            }
         }
 
         return $envelopeToSend;
     }
 
+    private function convertDeliveryDelayToMilliseconds(mixed $deliveryDelay, int $messageTimestamp): ?int
+    {
+        if ($deliveryDelay instanceof DateTimeInterface) {
+            $deliveryDelay = DatePoint::createFromInterface($deliveryDelay)->durationSince(DatePoint::createFromTimestamp($messageTimestamp));
+        }
+
+        if ($deliveryDelay instanceof Duration) {
+            $deliveryDelay = $deliveryDelay->inMilliseconds();
+        }
+
+        if ($deliveryDelay instanceof TimeSpan) {
+            $deliveryDelay = $deliveryDelay->toMilliseconds();
+        }
+
+        if ($deliveryDelay !== null && $deliveryDelay < 0) {
+            return null;
+        }
+
+        return $deliveryDelay;
+    }
+
     public function convertFromSymfonyMessage(Envelope $symfonyEnvelope, TransportInterface $symfonyTransport): Message
     {
         $headers = $symfonyEnvelope->last(MetadataStamp::class)->getMetadata();