Merge pull request #36 from kdambekalns/2.0

BUGFIX: Pass named arguments to executeCommand

Author: dimaip

Classes/Command/JobCommandController.php

@@ -154,7 +154,7 @@ public function executeCommand($queue, $messageCacheIdentifier)
         }
 
         /** @var Message $message */
-        $message = unserialize($this->messageCache->get($messageCacheIdentifier));
+        $message = $this->messageCache->get($messageCacheIdentifier);
         $queue = $this->queueManager->getQueue($queue);
         $this->jobManager->executeJobForMessage($queue, $message);
     }

Classes/Job/JobManager.php

@@ -100,8 +100,8 @@ public function waitAndExecute($queueName, $timeout = null)
         try {
             if (isset($queueSettings['executeIsolated']) && $queueSettings['executeIsolated'] === true) {
                 $messageCacheIdentifier = sha1(serialize($message));
-                $this->messageCache->set($messageCacheIdentifier, serialize($message));
-                Scripts::executeCommand('flowpack.jobqueue.common:job:execute', $this->flowSettings, false, [$queue->getName(), $messageCacheIdentifier]);
+                $this->messageCache->set($messageCacheIdentifier, $message);
+                Scripts::executeCommand('flowpack.jobqueue.common:job:execute', $this->flowSettings, false, ['queue' => $queue->getName(), 'messageCacheIdentifier' => $messageCacheIdentifier]);
                 $this->messageCache->remove($messageCacheIdentifier);
             } else {
                 $this->executeJobForMessage($queue, $message);

Classes/Queue/FakeQueue.php

@@ -12,6 +12,7 @@
  */
 
 use Neos\Flow\Annotations as Flow;
+use Neos\Cache\Frontend\VariableFrontend;
 use Neos\Flow\Core\Booting\Scripts;
 use Neos\Flow\Utility\Algorithms;
 
@@ -38,6 +39,12 @@ class FakeQueue implements QueueInterface
      */
     protected $async = false;
 
+    /**
+     * @Flow\Inject
+     * @var VariableFrontend
+     */
+    protected $messageCache;
+
     /**
      * @param string $name
      * @param array $options
@@ -76,7 +83,11 @@ public function submit($payload, array $options = [])
     {
         $messageId = Algorithms::generateUUID();
         $message = new Message($messageId, $payload);
-        $commandArguments = [$this->name, base64_encode(serialize($message))];
+
+        $messageCacheIdentifier = sha1(serialize($message));
+        $this->messageCache->set($messageCacheIdentifier, $message);
+        $commandArguments = ['queue' => $this->name, 'messageCacheIdentifier' => $messageCacheIdentifier];
+
         if ($this->async) {
             if (!method_exists(Scripts::class, 'executeCommandAsync')) {
                 throw new \RuntimeException('The "async" flag is set, but the currently used Flow version doesn\'t support this (Flow 3.3+ is required)', 1469116604);
@@ -149,7 +160,7 @@ public function count()
      */
     public function flush()
     {
-        //
+        // The FakeQueue does not support message flushing
     }
 
 }