Add handling of delay messages during initial consume and handling

Author: Jason Silva

Core/Handlers/MessageHandler.php

@@ -9,6 +9,7 @@
 use Smartbox\Integration\FrameworkBundle\Core\Exchange;
 use Smartbox\Integration\FrameworkBundle\Core\Messages\Context;
 use Smartbox\Integration\FrameworkBundle\Core\Messages\DeferredExchangeEnvelope;
+use Smartbox\Integration\FrameworkBundle\Core\Messages\DelayedExchangeEnvelope;
 use Smartbox\Integration\FrameworkBundle\Core\Messages\ErrorExchangeEnvelope;
 use Smartbox\Integration\FrameworkBundle\Core\Messages\ExchangeEnvelope;
 use Smartbox\Integration\FrameworkBundle\Core\Messages\FailedExchangeEnvelope;
@@ -408,7 +409,11 @@ public function onHandleException(
 
             } // If it's an exchange that can be retried later but it's failing due to an error
             elseif ($originalException instanceof DelayException) {
-                throw new UnrecoverableRestException('Delay Interceptor triggered.');
+                $delayPeriod = $exchangeBackup->getIn()->getHeader('delay');
+                $delayExchangeEnvelope = new DelayedExchangeEnvelope($exchangeBackup, $delayPeriod);
+
+                $fromQueue = $exchangeBackup->getHeader('from');
+                $this->deferExchangeMessage($delayExchangeEnvelope, $fromQueue);
             }
             elseif ($originalException instanceof RecoverableExceptionInterface && $retries < $this->retriesMax) {
 
@@ -548,6 +553,10 @@ public function handle(MessageInterface $message, EndpointInterface $endpointFro
 
                     return;
                 }
+            } elseif ($message instanceof DelayedExchangeEnvelope) {
+                $headers = $message->getBody()->getIn()->getHeaders();
+                unset($headers['delay']);
+                $message->getBody()->getIn()->setHeaders($headers);
             }
         }
         // Otherwise create the exchange

Core/Messages/DelayedExchangeEnvelope.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace Smartbox\Integration\FrameworkBundle\Core\Messages;
+
+use Smartbox\CoreBundle\Type\SerializableArray;
+use Smartbox\Integration\FrameworkBundle\Core\Exchange;
+
+class DelayedExchangeEnvelope extends ExchangeEnvelope
+{
+    const HEADER_DELAY_PERIOD = 'delay_period';
+
+    public function __construct(Exchange $exchange, int $delayPeriod)
+    {
+        parent::__construct($exchange);
+
+        $this->setHeader(self::HEADER_DELAY_PERIOD, $delayPeriod);
+    }
+
+}

Core/Processors/ControlFlow/DelayInterceptor.php

@@ -5,8 +5,6 @@
 use Smartbox\CoreBundle\Type\SerializableArray;
 use Smartbox\Integration\FrameworkBundle\Core\Exchange;
 use Smartbox\Integration\FrameworkBundle\Core\Processors\Exceptions\DelayException;
-use Smartbox\Integration\FrameworkBundle\Core\Processors\Exceptions\ThrottledException;
-use Smartbox\Integration\FrameworkBundle\Core\Processors\Exceptions\ThrottlingLimitReachedException;
 use Smartbox\Integration\FrameworkBundle\Core\Processors\Processor;
 
 class DelayInterceptor extends Processor
@@ -49,12 +47,13 @@ public function setDelayPeriod(int $delayPeriod)
      * @param Exchange          $exchange
      * @param SerializableArray $processingContext
      *
-     * @throws ThrottlingLimitReachedException
-     * @throws ThrottledException
+     * @throws DelayException
      */
     protected function doProcess(Exchange $exchange, SerializableArray $processingContext)
     {
-        if ($this->delayPeriod > 0 || $exchange->getIn()->getHeader('delay') > 0) {
+        $delayPeriodInSeconds = $exchange->getIn()->getHeader('delay') ?? null;
+
+        if (!is_null($delayPeriodInSeconds)) {
             throw new DelayException();
         }
     }

Tests/Unit/Core/Messages/DelayedExchangeEnvelopeTest.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Smartbox\Integration\FrameworkBundle\Tests\Unit\Core\Messages;
+
+use Smartbox\CoreBundle\Type\SerializableArray;
+use Smartbox\Integration\FrameworkBundle\Core\Exchange;
+use Smartbox\Integration\FrameworkBundle\Core\Messages\Context;
+use Smartbox\Integration\FrameworkBundle\Core\Messages\Message;
+use Smartbox\Integration\FrameworkBundle\Core\Messages\MessageInterface;
+use Smartbox\Integration\FrameworkBundle\Core\Messages\DelayedExchangeEnvelope;
+
+class DelayedExchangeEnvelopeTest extends \PHPUnit\Framework\TestCase
+{
+    /** @var DelayedExchangeEnvelope */
+    private $delayedExchangeEnvelope;
+
+    protected function setUp()
+    {
+        /** @var Context|\PHPUnit_Framework_MockObject_MockObject $context */
+        $context = $this->createMock(Context::class);
+
+        /** @var MessageInterface|\PHPUnit_Framework_MockObject_MockObject $message */
+        $message = $this->createMock(MessageInterface::class);
+        $message
+            ->expects($this->once())
+            ->method('getContext')
+            ->will($this->returnValue($context));
+
+        $exchange = new Exchange($message);
+
+        $this->delayedExchangeEnvelope = new DelayedExchangeEnvelope($exchange);
+    }
+
+    protected function tearDown()
+    {
+        $this->delayedExchangeEnvelope = null;
+    }
+
+    public function testConstruct()
+    {
+        $context = new Context();
+        $message = new Message(null, [], $context);
+        $exchange = new Exchange($message);
+        $delayPeriod = 10;
+
+        $delayedExchangeEnvelope = new DelayedExchangeEnvelope($exchange, $delayPeriod);
+
+        $this->assertSame($exchange, $delayedExchangeEnvelope->getExchange());
+        $this->assertSame($delayPeriod, $delayedExchangeEnvelope->getHeader(DelayedExchangeEnvelope::HEADER_DELAY_PERIOD));
+    }
+}