[consumption] Remove onInterrupted. Add onEnd. Remove context object and consumption interrupted exception.

Author: makasim

pkg/enqueue/Client/ConsumptionExtension/FlushSpoolProducerExtension.php

@@ -3,7 +3,7 @@
 namespace Enqueue\Client\ConsumptionExtension;
 
 use Enqueue\Client\SpoolProducer;
-use Enqueue\Consumption\Context;
+use Enqueue\Consumption\Context\End;
 use Enqueue\Consumption\Context\PostMessageReceived;
 use Enqueue\Consumption\EmptyExtensionTrait;
 use Enqueue\Consumption\ExtensionInterface;
@@ -30,7 +30,7 @@ public function onPostMessageReceived(PostMessageReceived $context): void
         $this->producer->flush();
     }
 
-    public function onInterrupted(Context $context)
+    public function onEnd(End $context): void
     {
         $this->producer->flush();
     }

pkg/enqueue/Consumption/ChainExtension.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Consumption;
 
+use Enqueue\Consumption\Context\End;
 use Enqueue\Consumption\Context\MessageReceived;
 use Enqueue\Consumption\Context\MessageResult;
 use Enqueue\Consumption\Context\PostConsume;
@@ -87,10 +88,10 @@ public function onPostConsume(PostConsume $context): void
         }
     }
 
-    public function onInterrupted(Context $context)
+    public function onEnd(End $context): void
     {
         foreach ($this->extensions as $extension) {
-            $extension->onInterrupted($context);
+            $extension->onEnd($context);
         }
     }
 }

pkg/enqueue/Consumption/Context.php

@@ -0,0 +1,63 @@
+<?php
+
+namespace Enqueue\Consumption\Context;
+
+use Interop\Queue\Context;
+use Psr\Log\LoggerInterface;
+
+final class End
+{
+    /**
+     * @var Context
+     */
+    private $context;
+
+    /**
+     * @var int
+     */
+    private $startTime;
+
+    /**
+     * @var int
+     */
+    private $endTime;
+
+    /**
+     * @var LoggerInterface
+     */
+    private $logger;
+
+    public function __construct(Context $context, int $startTime, int $endTime, LoggerInterface $logger)
+    {
+        $this->context = $context;
+        $this->logger = $logger;
+        $this->startTime = $startTime;
+        $this->endTime = $endTime;
+    }
+
+    public function getContext(): Context
+    {
+        return $this->context;
+    }
+
+    public function getLogger(): LoggerInterface
+    {
+        return $this->logger;
+    }
+
+    /**
+     * In milliseconds.
+     */
+    public function getStartTime(): int
+    {
+        return $this->startTime;
+    }
+
+    /**
+     * In milliseconds.
+     */
+    public function getEndTime(): int
+    {
+        return $this->startTime;
+    }
+}

pkg/enqueue/Consumption/Context/End.php

@@ -2,6 +2,7 @@
 
 namespace Enqueue\Consumption;
 
+use Enqueue\Consumption\Context\End;
 use Enqueue\Consumption\Context\MessageReceived;
 use Enqueue\Consumption\Context\MessageResult;
 use Enqueue\Consumption\Context\PostConsume;
@@ -45,7 +46,7 @@ public function onPostConsume(PostConsume $context): void
     {
     }
 
-    public function onInterrupted(Context $context)
+    public function onEnd(End $context): void
     {
     }
 }

pkg/enqueue/Consumption/EmptyExtensionTrait.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace Enqueue\Consumption;
+
+use Enqueue\Consumption\Context\End;
+
+interface EndExtensionInterface
+{
+    /**
+     * Executed only once just before QueueConsumer::consume returns.
+     */
+    public function onEnd(End $context): void;
+}

pkg/enqueue/Consumption/EndExtensionInterface.php

@@ -2,13 +2,6 @@
 
 namespace Enqueue\Consumption;
 
-interface ExtensionInterface extends StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, MessageReceivedExtensionInterface, PostMessageReceivedExtensionInterface, MessageResultExtensionInterface, ProcessorExceptionExtensionInterface, PostConsumeExtensionInterface
+interface ExtensionInterface extends StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, MessageReceivedExtensionInterface, PostMessageReceivedExtensionInterface, MessageResultExtensionInterface, ProcessorExceptionExtensionInterface, PostConsumeExtensionInterface, EndExtensionInterface
 {
-    /**
-     * Called when the consumption was interrupted by an extension or exception
-     * In case of exception it will be present in the context.
-     *
-     * @param Context $context
-     */
-    public function onInterrupted(Context $context);
 }