[consumption] Add ability to change process exit status from within queue consumer extension

pkg/enqueue/Consumption/Context/End.php

@@ -27,12 +27,23 @@ final class End
      */
     private $logger;
 
-    public function __construct(Context $context, int $startTime, int $endTime, LoggerInterface $logger)
-    {
+    /**
+     * @var int
+     */
+    private $exitStatus;
+
+    public function __construct(
+        Context $context,
+        int $startTime,
+        int $endTime,
+        LoggerInterface $logger,
+        ?int $exitStatus = null
+    ) {
         $this->context = $context;
         $this->logger = $logger;
         $this->startTime = $startTime;
         $this->endTime = $endTime;
+        $this->exitStatus = $exitStatus;
     }
 
     public function getContext(): Context
@@ -60,4 +71,9 @@ public function getEndTime(): int
     {
         return $this->startTime;
     }
+
+    public function getExitStatus(): ?int
+    {
+        return $this->exitStatus;
+    }
 }

pkg/enqueue/Consumption/Context/PostConsume.php

@@ -43,6 +43,11 @@ final class PostConsume
      */
     private $executionInterrupted;
 
+    /**
+     * @var int
+     */
+    private $exitStatus;
+
     public function __construct(Context $context, SubscriptionConsumer $subscriptionConsumer, int $receivedMessagesCount, int $cycle, int $startTime, LoggerInterface $logger)
     {
         $this->context = $context;
@@ -85,13 +90,19 @@ public function getLogger(): LoggerInterface
         return $this->logger;
     }
 
+    public function getExitStatus(): ?int
+    {
+        return $this->exitStatus;
+    }
+
     public function isExecutionInterrupted(): bool
     {
         return $this->executionInterrupted;
     }
 
-    public function interruptExecution(): void
+    public function interruptExecution(?int $exitStatus = null): void
     {
+        $this->exitStatus = $exitStatus;
         $this->executionInterrupted = true;
     }
 }

pkg/enqueue/Consumption/Context/PostMessageReceived.php

@@ -45,6 +45,11 @@ final class PostMessageReceived
      */
     private $executionInterrupted;
 
+    /**
+     * @var int
+     */
+    private $exitStatus;
+
     public function __construct(
         Context $context,
         Consumer $consumer,
@@ -96,13 +101,19 @@ public function getResult()
         return $this->result;
     }
 
+    public function getExitStatus(): ?int
+    {
+        return $this->exitStatus;
+    }
+
     public function isExecutionInterrupted(): bool
     {
         return $this->executionInterrupted;
     }
 
-    public function interruptExecution(): void
+    public function interruptExecution(?int $exitStatus = null): void
     {
+        $this->exitStatus = $exitStatus;
         $this->executionInterrupted = true;
     }
 }

pkg/enqueue/Consumption/Context/PreConsume.php

@@ -43,6 +43,11 @@ final class PreConsume
      */
     private $executionInterrupted;
 
+    /**
+     * @var int
+     */
+    private $exitStatus;
+
     public function __construct(Context $context, SubscriptionConsumer $subscriptionConsumer, LoggerInterface $logger, int $cycle, int $receiveTimeout, int $startTime)
     {
         $this->context = $context;
@@ -85,13 +90,19 @@ public function getStartTime(): int
         return $this->startTime;
     }
 
+    public function getExitStatus(): ?int
+    {
+        return $this->exitStatus;
+    }
+
     public function isExecutionInterrupted(): bool
     {
         return $this->executionInterrupted;
     }
 
-    public function interruptExecution(): void
+    public function interruptExecution(?int $exitStatus = null): void
     {
+        $this->exitStatus = $exitStatus;
         $this->executionInterrupted = true;
     }
 }

pkg/enqueue/Consumption/Context/Start.php

@@ -38,6 +38,11 @@ final class Start
      */
     private $executionInterrupted;
 
+    /**
+     * @var int
+     */
+    private $exitStatus;
+
     /**
      * @param BoundProcessor[] $processors
      */
@@ -105,13 +110,19 @@ public function changeBoundProcessors(array $processors): void
         });
     }
 
+    public function getExitStatus(): ?int
+    {
+        return $this->exitStatus;
+    }
+
     public function isExecutionInterrupted(): bool
     {
         return $this->executionInterrupted;
     }
 
-    public function interruptExecution(): void
+    public function interruptExecution(?int $exitStatus = null): void
     {
+        $this->exitStatus = $exitStatus;
         $this->executionInterrupted = true;
     }
 }

pkg/enqueue/Consumption/Extension/CaptureExitStatusExtension.php

@@ -0,0 +1,35 @@
+<?php
+
+namespace Enqueue\Consumption\Extension;
+
+use Enqueue\Consumption\Context\End;
+use Enqueue\Consumption\EndExtensionInterface;
+
+class CaptureExitStatusExtension implements EndExtensionInterface
+{
+    /**
+     * @var int
+     */
+    private $exitStatus;
+
+    /**
+     * @var bool
+     */
+    private $isExitStatusCaptured = false;
+
+    public function onEnd(End $context): void
+    {
+        $this->exitStatus = $context->getExitStatus();
+        $this->isExitStatusCaptured = true;
+    }
+
+    public function getExitStatus(): ?int
+    {
+        return $this->exitStatus;
+    }
+
+    public function isExitStatusCaptured(): bool
+    {
+        return $this->isExitStatusCaptured;
+    }
+}

pkg/enqueue/Consumption/QueueConsumer.php

@@ -147,7 +147,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
         $extension->onStart($start);
 
         if ($start->isExecutionInterrupted()) {
-            $this->onEnd($extension, $startTime);
+            $this->onEnd($extension, $startTime, $start->getExitStatus());
 
             return;
         }
@@ -256,7 +256,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
             $extension->onPreConsume($preConsume);
 
             if ($preConsume->isExecutionInterrupted()) {
-                $this->onEnd($extension, $startTime, $subscriptionConsumer);
+                $this->onEnd($extension, $startTime, $preConsume->getExitStatus(), $subscriptionConsumer);
 
                 return;
             }
@@ -267,7 +267,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
             $extension->onPostConsume($postConsume);
 
             if ($interruptExecution || $postConsume->isExecutionInterrupted()) {
-                $this->onEnd($extension, $startTime, $subscriptionConsumer);
+                $this->onEnd($extension, $startTime, $postConsume->getExitStatus(), $subscriptionConsumer);
 
                 return;
             }
@@ -286,11 +286,12 @@ public function setFallbackSubscriptionConsumer(SubscriptionConsumer $fallbackSu
         $this->fallbackSubscriptionConsumer = $fallbackSubscriptionConsumer;
     }
 
-    private function onEnd(ExtensionInterface $extension, int $startTime, SubscriptionConsumer $subscriptionConsumer = null): void
+    private function onEnd(ExtensionInterface $extension, int $startTime, ?int $exitStatus = null, SubscriptionConsumer $subscriptionConsumer = null): void
     {
         $endTime = (int) (microtime(true) * 1000);
 
-        $extension->onEnd(new End($this->interopContext, $startTime, $endTime, $this->logger));
+        $endContext = new End($this->interopContext, $startTime, $endTime, $this->logger, $exitStatus);
+        $extension->onEnd($endContext);
 
         if ($subscriptionConsumer) {
             $subscriptionConsumer->unsubscribeAll();

pkg/enqueue/Symfony/Consumption/ConsumeCommand.php

@@ -3,6 +3,7 @@
 namespace Enqueue\Symfony\Consumption;
 
 use Enqueue\Consumption\ChainExtension;
+use Enqueue\Consumption\Extension\CaptureExitStatusExtension;
 use Enqueue\Consumption\QueueConsumerInterface;
 use Psr\Container\ContainerInterface;
 use Psr\Container\NotFoundExceptionInterface;
@@ -75,9 +76,12 @@ protected function execute(InputInterface $input, OutputInterface $output): ?int
             array_unshift($extensions, $loggerExtension);
         }
 
+        $captureExitStatusExtension = new CaptureExitStatusExtension();
+        array_unshift($extensions, $captureExitStatusExtension);
+
         $consumer->consume(new ChainExtension($extensions));
 
-        return null;
+        return $captureExitStatusExtension->getExitStatus();
     }
 
     private function getQueueConsumer(string $name): QueueConsumerInterface

pkg/enqueue/Tests/Consumption/QueueConsumerTest.php

@@ -16,6 +16,7 @@
 use Enqueue\Consumption\Context\ProcessorException;
 use Enqueue\Consumption\Context\Start;
 use Enqueue\Consumption\Exception\InvalidArgumentException;
+use Enqueue\Consumption\Extension\CaptureExitStatusExtension;
 use Enqueue\Consumption\ExtensionInterface;
 use Enqueue\Consumption\QueueConsumer;
 use Enqueue\Consumption\Result;
@@ -1429,6 +1430,30 @@ public function testShouldCallProcessorAsMessageComeAlong()
         $this->assertSame($fooConsumerStub, $actualContexts[2]->getConsumer());
     }
 
+    public function testCaptureExitStatus()
+    {
+        $testExitCode = 5;
+
+        $stubExtension = $this->createExtension();
+
+        $stubExtension
+            ->expects($this->once())
+            ->method('onStart')
+            ->with($this->isInstanceOf(Start::class))
+            ->willReturnCallback(function (Start $context) use ($testExitCode) {
+                $context->interruptExecution($testExitCode);
+            })
+        ;
+
+        $exitExtension = new CaptureExitStatusExtension();
+
+        $consumer = new QueueConsumer($this->createContextStub(), $stubExtension);
+        $consumer->consume(new ChainExtension([$exitExtension]));
+
+        $this->assertEquals($testExitCode, $exitExtension->getExitStatus());
+        $this->assertTrue($exitExtension->isExitStatusCaptured());
+    }
+
     /**
      * @return \PHPUnit_Framework_MockObject_MockObject
      */