Catch not found cache item

app/bundles/CoreBundle/Helper/CacheStorageHelper.php

@@ -101,6 +101,14 @@ public function __construct($adaptor, $namespace = null, Connection $connection
         $this->setCacheAdaptor();
     }
 
+    /**
+     * @return string|false
+     */
+    public function getAdaptorClassName()
+    {
+        return get_class($this->cacheAdaptor);
+    }
+
     /**
      * @param $name
      * @param $data

app/bundles/EmailBundle/Config/config.php

@@ -127,6 +127,7 @@
                     'mautic.transport.momentum.callback',
                     'mautic.queue.service',
                     'mautic.email.helper.request.storage',
+                    'monolog.logger.mautic',
                 ],
             ],
             'mautic.email.monitored.bounce.subscriber' => [

app/bundles/EmailBundle/EventListener/MomentumSubscriber.php

@@ -22,6 +22,7 @@
 use Mautic\QueueBundle\Queue\QueueName;
 use Mautic\QueueBundle\Queue\QueueService;
 use Mautic\QueueBundle\QueueEvents;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\HttpFoundation\Request;
 
 /**
@@ -48,15 +49,18 @@ class MomentumSubscriber extends CommonSubscriber
      * @param MomentumCallbackInterface $momentumCallback
      * @param QueueService              $queueService
      * @param RequestStorageHelper      $requestStorageHelper
+     * @param LoggerInterface           $logger
      */
     public function __construct(
         MomentumCallbackInterface $momentumCallback,
         QueueService $queueService,
-        RequestStorageHelper $requestStorageHelper
+        RequestStorageHelper $requestStorageHelper,
+        LoggerInterface $logger
     ) {
         $this->momentumCallback     = $momentumCallback;
         $this->queueService         = $queueService;
         $this->requestStorageHelper = $requestStorageHelper;
+        $this->logger               = $logger;
     }
 
     /**
@@ -80,9 +84,15 @@ public function onMomentumWebhookQueueProcessing(QueueConsumerEvent $event)
         if ($event->checkTransport(MomentumTransport::class)) {
             $payload = $event->getPayload();
             $key     = $payload['key'];
-            $request = $this->requestStorageHelper->getRequest($key);
-            $this->momentumCallback->processCallbackRequest($request);
-            $this->requestStorageHelper->deleteCachedRequest($key);
+
+            try {
+                $request = $this->requestStorageHelper->getRequest($key);
+                $this->momentumCallback->processCallbackRequest($request);
+                $this->requestStorageHelper->deleteCachedRequest($key);
+            } catch (\UnexpectedValueException $e) {
+                $this->logger->error($e->getMessage());
+            }
+
             $event->setResult(QueueConsumerResults::ACKNOWLEDGE);
         }
     }
@@ -96,7 +106,7 @@ public function onMomentumWebhookRequest(TransportWebhookEvent $event)
         if ($this->queueService->isQueueEnabled() && $event->transportIsInstanceOf($transport)) {
             // Beanstalk jobs are limited to 65,535 kB. Momentum can send up to 10.000 items per request.
             // One item has about 1,6 kB. Lets store the request to the cache storage instead of the job itself.
-            $key       = $this->requestStorageHelper->storeRequest($transport, $event->getRequest());
+            $key = $this->requestStorageHelper->storeRequest($transport, $event->getRequest());
             $this->queueService->publishToQueue(QueueName::TRANSPORT_WEBHOOK, ['transport' => $transport, 'key' => $key]);
             $event->stopPropagation();
         }

app/bundles/EmailBundle/Helper/RequestStorageHelper.php

@@ -60,17 +60,28 @@ public function storeRequest($transportName, Request $request)
      * @param string $key
      *
      * @return Request
+     *
+     * @throws \UnexpectedValueException
      */
     public function getRequest($key)
     {
-        return new Request([], $this->cacheStorage->get($key));
+        $key           = $this->removeCachePrefix($key);
+        $cachedRequest = $this->cacheStorage->get($key);
+
+        if (false === $cachedRequest) {
+            throw new \UnexpectedValueException("Request with key '{$key}' was not found in the cache store '{$this->cacheStorage->getAdaptorClassName()}'.");
+        }
+
+        return new Request([], $cachedRequest);
     }
 
     /**
      * @param string $key
      */
     public function deleteCachedRequest($key)
     {
+        $key = $this->removeCachePrefix($key);
+
         $this->cacheStorage->delete($key);
     }
 
@@ -83,11 +94,33 @@ public function deleteCachedRequest($key)
      */
     public function getTransportNameFromKey($key)
     {
-        list($transportName) = explode(self::KEY_SEPARATOR, $key);
+        $key = $this->removeCachePrefix($key);
+
+        // Take the part before the key separator as the serialized transpot name.
+        list($serializedTransportName) = explode(self::KEY_SEPARATOR, $key);
+
+        // Unserialize transport name to the standard full class name.
+        $transportName = str_replace('|', '\\', $serializedTransportName);
 
         return $transportName;
     }
 
+    /**
+     * Remove the default cache key prefix if set.
+     *
+     * @param string $key
+     *
+     * @return string
+     */
+    private function removeCachePrefix($key)
+    {
+        if (strpos($key, 'mautic:') === 0) {
+            $key = ltrim($key, 'mautic:');
+        }
+
+        return $key;
+    }
+
     /**
      * Generates unique hash in format $transportName:webhook_request:unique.hash.
      *

app/bundles/EmailBundle/Tests/EventListener/MomentumSubscriberTest.php

@@ -20,12 +20,16 @@
 use Mautic\QueueBundle\Queue\QueueConsumerResults;
 use Mautic\QueueBundle\Queue\QueueName;
 use Mautic\QueueBundle\Queue\QueueService;
+use Psr\Log\LoggerInterface;
 use Symfony\Component\HttpFoundation\Request;
 
 class MomentumSubscriberTest extends \PHPUnit_Framework_TestCase
 {
     private $queueServiceMock;
     private $momentumCallbackMock;
+    private $requestStorageHelperMock;
+    private $loggerMock;
+    private $momentumSubscriber;
 
     protected function setUp()
     {
@@ -34,7 +38,13 @@ protected function setUp()
         $this->momentumCallbackMock     = $this->createMock(MomentumCallbackInterface::class);
         $this->queueServiceMock         = $this->createMock(QueueService::class);
         $this->requestStorageHelperMock = $this->createMock(RequestStorageHelper::class);
-        $this->momentumSubscriber       = new MomentumSubscriber($this->momentumCallbackMock, $this->queueServiceMock, $this->requestStorageHelperMock);
+        $this->loggerMock               = $this->createMock(LoggerInterface::class);
+        $this->momentumSubscriber       = new MomentumSubscriber(
+            $this->momentumCallbackMock,
+            $this->queueServiceMock,
+            $this->requestStorageHelperMock,
+            $this->loggerMock
+        );
     }
 
     public function testOnMomentumWebhookQueueProcessingForNonMomentumTransport()
@@ -95,6 +105,41 @@ public function testOnMomentumWebhookQueueProcessingForMomentumTransport()
         $this->momentumSubscriber->onMomentumWebhookQueueProcessing($queueConsumerEvent);
     }
 
+    public function testOnMomentumWebhookQueueProcessingForMomentumTransportIfRequestNotFounc()
+    {
+        $queueConsumerEvent = $this->createMock(QueueConsumerEvent::class);
+
+        $queueConsumerEvent->expects($this->once())
+            ->method('getPayload')
+            ->willReturn([
+                'transport' => MomentumTransport::class,
+                'key'       => 'value',
+            ]);
+
+        $queueConsumerEvent->expects($this->once())
+            ->method('checkTransport')
+            ->with(MomentumTransport::class)
+            ->willReturn(true);
+
+        $this->requestStorageHelperMock->expects($this->once())
+            ->method('getRequest')
+            ->with('value')
+            ->will($this->throwException(new \UnexpectedValueException('Error message')));
+
+        $this->momentumCallbackMock->expects($this->never())
+            ->method('processCallbackRequest');
+
+        $this->loggerMock->expects($this->once())
+            ->method('error')
+            ->with('Error message');
+
+        $queueConsumerEvent->expects($this->once())
+            ->method('setResult')
+            ->with(QueueConsumerResults::ACKNOWLEDGE);
+
+        $this->momentumSubscriber->onMomentumWebhookQueueProcessing($queueConsumerEvent);
+    }
+
     public function testOnMomentumWebhookRequestWhenQueueIsDisabled()
     {
         $transportWebhookEvent = $this->createMock(TransportWebhookEvent::class);

app/bundles/EmailBundle/Tests/Helper/RequestStorageHelperTest.php

@@ -72,8 +72,30 @@ public function testGetRequest()
         $this->assertEquals($payload, $request->request->all());
     }
 
+    public function testGetRequestIfNotFound()
+    {
+        $payload = ['some' => 'values'];
+        $key     = MomentumTransport::class.':webhook_request:5b43832134cfb0.36545510';
+
+        $this->cacheStorageMock->expects($this->once())
+            ->method('get')
+            ->with($key)
+            ->willReturn(false);
+
+        $this->expectException(\UnexpectedValueException::class);
+        $this->helper->getRequest($key);
+    }
+
     public function testGetTransportNameFromKey()
     {
         $this->assertEquals(MomentumTransport::class, $this->helper->getTransportNameFromKey('Mautic\EmailBundle\Swiftmailer\Transport\MomentumTransport:webhook_request:5b43832134cfb0.36545510'));
     }
+
+    /**
+     * The StorageHelper will add '%mautic.db_table_prefix%' as a prefix to each cache key.
+     */
+    public function testGetTransportNameFromKeyWithGlobalPrefix()
+    {
+        $this->assertEquals(MomentumTransport::class, $this->helper->getTransportNameFromKey('mautic:Mautic|EmailBundle|Swiftmailer|Transport|MomentumTransport:webhook_request:5bfbe8ce671198.00044461'));
+    }
 }