Merge pull request #33 from Flowpack/bugfix/remove-message-cache-on-failure

BUGFIX: Remove message cache file on failure

Author: kdambekalns

Classes/Job/JobManager.php

@@ -87,6 +87,8 @@ public function queue(string $queueName, JobInterface $job, array $options = [])
      */
     public function waitAndExecute(string $queueName, $timeout = null): ?Message
     {
+
+        $messageCacheIdentifier = null;
         $queue = $this->queueManager->getQueue($queueName);
         $message = $queue->waitAndReserve($timeout);
         if ($message === null) {
@@ -102,7 +104,6 @@ public function waitAndExecute(string $queueName, $timeout = null): ?Message
                 $messageCacheIdentifier = sha1(serialize($message));
                 $this->messageCache->set($messageCacheIdentifier, $message);
                 Scripts::executeCommand('flowpack.jobqueue.common:job:execute', $this->flowSettings, false, [$queue->getName(), $messageCacheIdentifier]);
-                $this->messageCache->remove($messageCacheIdentifier);
             } else {
                 $this->executeJobForMessage($queue, $message);
             }
@@ -118,6 +119,10 @@ public function waitAndExecute(string $queueName, $timeout = null): ?Message
                 $this->emitMessageFailed($queue, $message, $exception);
                 throw new JobQueueException(sprintf('Job execution for job (message: "%s", queue: "%s") failed (%d/%d trials) - ABORTING', $message->getIdentifier(), $queue->getName(), $message->getNumberOfReleases() + 1, $maximumNumberOfReleases + 1), 1334056584, $exception);
             }
+        } finally {
+            if ($messageCacheIdentifier !== null) {
+                $this->messageCache->remove($messageCacheIdentifier);
+            }
         }
 
         $queue->finish($message->getIdentifier());