TASK: Dispatch PreRejectEvent

.ddev/providers/acquia.yaml

@@ -45,7 +45,7 @@ db_pull_command:
     # just using `acli pull:db ${ACQUIA_ENVIRONMENT_ID}`
     echo "Using ACQUIA_ENVIRONMENT_ID=${ACQUIA_ENVIRONMENT_ID}"
     set -x   # You can enable bash debugging output by uncommenting
-    db_dump=$(acli pull:db ${ACQUIA_ENVIRONMENT_ID} --no-interaction --no-import | tail -2l | xargs)
+    db_dump=$(acli pull:db ${ACQUIA_ENVIRONMENT_ID} --no-interaction --no-import | tail -2l | xargs |  sed 's/^.* //')
     ls /var/www/html/.ddev >/dev/null # This just refreshes stale NFS if possible
     cp ${db_dump} /var/www/html/.ddev/.downloads/db.sql.gz
 

Classes/Event/PreRejectEvent.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the "t3_messenger" Extension for TYPO3 CMS.
+ *
+ * For the full copyright and license information, please read the
+ * LICENSE.txt file that was distributed with this source code.
+ */
+
+namespace Ssch\T3Messenger\Event;
+
+use Symfony\Component\Messenger\Envelope;
+
+final class PreRejectEvent
+{
+    private Envelope $envelope;
+
+    public function __construct(Envelope $envelope)
+    {
+        $this->envelope = $envelope;
+    }
+
+    public function getEnvelope(): Envelope
+    {
+        return $this->envelope;
+    }
+}

Classes/Transport/DoctrineTransportFactory.php

@@ -14,6 +14,7 @@
 use Doctrine\DBAL\DriverManager;
 use Doctrine\DBAL\Platforms\PostgreSQLPlatform;
 use Doctrine\DBAL\Types\Type;
+use Psr\EventDispatcher\EventDispatcherInterface;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\Connection;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\PostgreSqlConnection;
@@ -37,6 +38,13 @@ final class DoctrineTransportFactory implements TransportFactoryInterface
         SetType::TYPE => SetType::class,
     ];
 
+    private EventDispatcherInterface $eventDispatcher;
+
+    public function __construct(EventDispatcherInterface $eventDispatcher)
+    {
+        $this->eventDispatcher = $eventDispatcher;
+    }
+
     /**
      * @param array<mixed> $options
      */
@@ -64,8 +72,8 @@ public function createTransport(string $dsn, array $options, SerializerInterface
 
         return new DoctrineTransportWrapper(new DoctrineTransport(
             $connection,
-            $serializer
-        ), $connection->getConfiguration(), $driverConnection);
+            $serializer,
+        ), $connection->getConfiguration(), $driverConnection, $this->eventDispatcher);
     }
 
     /**

Classes/Transport/DoctrineTransportWrapper.php

@@ -15,6 +15,8 @@
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Schema\Table;
 use Doctrine\DBAL\Types\Types;
+use Psr\EventDispatcher\EventDispatcherInterface;
+use Ssch\T3Messenger\Event\PreRejectEvent;
 use Symfony\Component\Messenger\Bridge\Doctrine\Transport\DoctrineTransport;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
@@ -30,14 +32,18 @@ final class DoctrineTransportWrapper implements TransportInterface, SetupableTra
 
     private array $configuration;
 
+    private EventDispatcherInterface $eventDispatcher;
+
     public function __construct(
         DoctrineTransport $doctrineTransport,
         array $configuration,
-        DBALConnection $driverConnection
+        DBALConnection $driverConnection,
+        EventDispatcherInterface $eventDispatcher
     ) {
         $this->doctrineTransport = $doctrineTransport;
         $this->configuration = $configuration;
         $this->driverConnection = $driverConnection;
+        $this->eventDispatcher = $eventDispatcher;
     }
 
     public function setup(): void
@@ -75,6 +81,7 @@ public function ack(Envelope $envelope): void
 
     public function reject(Envelope $envelope): void
     {
+        $this->eventDispatcher->dispatch(new PreRejectEvent($envelope));
         $this->doctrineTransport->reject($envelope);
     }
 

Tests/Functional/Fixtures/Extensions/t3_messenger_test/Classes/EventListener/PreRejectEventListener.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the "t3_messenger" Extension for TYPO3 CMS.
+ *
+ * For the full copyright and license information, please read the
+ * LICENSE.txt file that was distributed with this source code.
+ */
+
+namespace Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\EventListener;
+
+use Ssch\T3Messenger\Event\PreRejectEvent;
+
+final class PreRejectEventListener
+{
+    /**
+     * @var PreRejectEvent[]
+     */
+    private array $events = [];
+
+    public function __invoke(PreRejectEvent $event): void
+    {
+        $this->events[] = $event;
+    }
+
+    public function getEvents(): array
+    {
+        return $this->events;
+    }
+}

Tests/Functional/Fixtures/Extensions/t3_messenger_test/Configuration/Services.php

@@ -9,8 +9,10 @@
  * LICENSE.txt file that was distributed with this source code.
  */
 
+use Ssch\T3Messenger\Event\PreRejectEvent;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Controller\MessengerController;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\EventListener\BeforeMailerSentMessageEventListener;
+use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\EventListener\PreRejectEventListener;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Handlers\MyMessengerHandler;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Service\MyService;
 use Symfony\Component\DependencyInjection\Loader\Configurator\ContainerConfigurator;
@@ -27,5 +29,8 @@
     $services->set(BeforeMailerSentMessageEventListener::class)->tag('event.listener', [
         'event' => BeforeMailerSentMessageEvent::class,
     ]);
+    $services->set(PreRejectEventListener::class)->tag('event.listener', [
+        'event' => PreRejectEvent::class,
+    ]);
     $services->set(MessengerController::class);
 };

Tests/Functional/MessengerTest.php

@@ -11,14 +11,18 @@
 
 namespace Ssch\T3Messenger\Tests\Functional;
 
+use Ssch\T3Messenger\Event\PreRejectEvent;
 use Ssch\T3Messenger\Exception\ValidationFailedException;
 use Ssch\T3Messenger\Stamp\SiteStamp;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Command\MyCommand;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Command\MyFailingCommand;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Command\MyOtherCommand;
+use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\EventListener\PreRejectEventListener;
 use Ssch\T3Messenger\Tests\Functional\Fixtures\Extensions\t3_messenger_test\Classes\Service\MyService;
+use Symfony\Component\EventDispatcher\EventDispatcher;
 use Symfony\Component\Messenger\EventListener\StopWorkerOnFailureLimitListener;
 use Symfony\Component\Messenger\Stamp\HandledStamp;
+use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp;
 use Symfony\Component\Messenger\Transport\TransportInterface;
 use Symfony\Component\Messenger\Worker;
 use TYPO3\CMS\Core\Http\Uri;
@@ -88,7 +92,7 @@ public function testThatFailingCommandIsTransferredToFailureTransport(): void
             'async' => $this->get('messenger.transport.async'),
         ];
 
-        /** @var \Symfony\Component\EventDispatcher\EventDispatcher $eventDispatcher */
+        /** @var EventDispatcher $eventDispatcher */
         $eventDispatcher = $this->get('event_dispatcher');
         $eventDispatcher->addSubscriber(new StopWorkerOnFailureLimitListener(1));
 
@@ -100,6 +104,38 @@ public function testThatFailingCommandIsTransferredToFailureTransport(): void
         self::assertCount(1, $transport->get());
     }
 
+    public function testThatOnRejectAnPreRejectEventIsDispatched(): void
+    {
+        $this->get(MyService::class)->dispatch(new MyFailingCommand('note'));
+
+        $receivers = [
+            'async' => $this->get('messenger.transport.async'),
+        ];
+
+        /** @var EventDispatcher $eventDispatcher */
+        $eventDispatcher = $this->get('event_dispatcher');
+        $eventDispatcher->addSubscriber(new StopWorkerOnFailureLimitListener(1));
+
+        $worker = new Worker($receivers, $this->get('command.bus'), $eventDispatcher);
+        $worker->run();
+
+        /** @var TransportInterface $transport */
+        $transport = $this->get('messenger.transport.failed');
+        foreach ($transport->get() as $message) {
+            $transport->reject($message);
+        }
+
+        $preRejectEventListener = $this->get(PreRejectEventListener::class);
+
+        $events = array_filter(
+            $preRejectEventListener->getEvents(),
+            fn (PreRejectEvent $event) => $event->getEnvelope()->last(SentToFailureTransportStamp::class) !== null
+        );
+
+        self::assertCount(1, $events);
+
+    }
+
     public function testThatServerRequestContextMiddlewareIsDefinedCorrectly(): void
     {
         $uri = new Uri($this->site->getBase()->__toString() . '/');