Catch symfony/messenger exceptions on message dispatch and convert it to job failure (#56)

Author: yann-eugone

src/DispatchMessageJobLauncher.php

@@ -4,6 +4,7 @@
 
 namespace Yokai\Batch\Bridge\Symfony\Messenger;
 
+use Symfony\Component\Messenger\Exception\ExceptionInterface;
 use Symfony\Component\Messenger\MessageBusInterface;
 use Yokai\Batch\BatchStatus;
 use Yokai\Batch\Factory\JobExecutionFactory;
@@ -35,8 +36,17 @@ public function launch(string $name, array $configuration = []): JobExecution
         $jobExecution->setStatus(BatchStatus::PENDING);
         $this->jobExecutionStorage->store($jobExecution);
 
-        // dispatch message
-        $this->messageBus->dispatch(new LaunchJobMessage($name, $configuration));
+        try {
+            // dispatch message
+            $this->messageBus->dispatch(new LaunchJobMessage($name, $configuration));
+        } catch (ExceptionInterface $exception) {
+            // if a messenger exception occurs, it will be converted to job failure
+            $jobExecution->setStatus(BatchStatus::FAILED);
+            $jobExecution->addFailureException($exception);
+            $this->jobExecutionStorage->store($jobExecution);
+
+            return $jobExecution;
+        }
 
         // re-fetch and return job execution from storage
         // if transport is synchronous, job execution may have been filled during execution

tests/DispatchMessageJobLauncherTest.php

@@ -8,6 +8,7 @@
 use Prophecy\Argument;
 use Prophecy\PhpUnit\ProphecyTrait;
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\MessageBusInterface;
 use Yokai\Batch\BatchStatus;
 use Yokai\Batch\Bridge\Symfony\Messenger\DispatchMessageJobLauncher;
@@ -81,4 +82,30 @@ static function ($message): bool {
         self::assertSame('123456789', $jobExecutionFromStorage->getId());
         self::assertSame(BatchStatus::PENDING, $jobExecutionFromStorage->getStatus()->getValue());
     }
+
+    public function testLaunchAndMessengerFail(): void
+    {
+        $messageBus = $this->prophesize(MessageBusInterface::class);
+        $messageBus->dispatch(Argument::any())
+            ->shouldBeCalled()
+            ->willThrow(new TransportException('This is a test'));
+
+        $jobLauncher = new DispatchMessageJobLauncher(
+            new JobExecutionFactory(new UniqidJobExecutionIdGenerator()),
+            $storage = new InMemoryJobExecutionStorage(),
+            $messageBus->reveal()
+        );
+
+        $jobExecutionFromLauncher = $jobLauncher->launch('testing');
+
+        [$jobExecutionFromStorage] = $storage->getExecutions();
+        self::assertSame($jobExecutionFromLauncher, $jobExecutionFromStorage);
+
+        self::assertSame('testing', $jobExecutionFromStorage->getJobName());
+        self::assertSame(BatchStatus::FAILED, $jobExecutionFromStorage->getStatus()->getValue());
+        self::assertCount(1, $jobExecutionFromStorage->getFailures());
+        $failure = $jobExecutionFromStorage->getFailures()[0];
+        self::assertSame(TransportException::class, $failure->getClass());
+        self::assertSame('This is a test', $failure->getMessage());
+    }
 }