RabbitMQ implementation

Author: mnapoli

src/MyCLabs/Work/Dispatcher/RabbitMQWorkDispatcher.php

@@ -0,0 +1,50 @@
+<?php
+
+namespace MyCLabs\Work\Dispatcher;
+
+use MyCLabs\Work\Task\Task;
+use PhpAmqpLib\Channel\AMQPChannel;
+use PhpAmqpLib\Message\AMQPMessage;
+
+/**
+ * RabbitMQ implementation.
+ *
+ * @author Matthieu Napoli <[email protected]>
+ */
+class RabbitMQWorkDispatcher implements WorkDispatcher
+{
+    /**
+     * @var AMQPChannel
+     */
+    private $channel;
+
+    /**
+     * @var string
+     */
+    private $queue;
+
+    /**
+     * @param AMQPChannel $channel
+     * @param string      $queue
+     */
+    public function __construct(AMQPChannel $channel, $queue)
+    {
+        $this->channel = $channel;
+        $this->queue = $queue;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function runBackground(Task $task)
+    {
+        $message = new AMQPMessage(
+            serialize($task),
+            [
+                'delivery_mode' => 2, // make message persistent
+            ]
+        );
+
+        $this->channel->basic_publish($message, '', $this->queue);
+    }
+}

src/MyCLabs/Work/Dispatcher/SimpleWorkDispatcher.php

@@ -32,6 +32,7 @@ public function __construct(SimpleWorker $simpleWorker)
 
     /**
      * {@inheritdoc}
+     * @return mixed This particular implementation can return the result since it's executed synchronously
      */
     public function runBackground(Task $task)
     {

src/MyCLabs/Work/Worker/RabbitMQWorker.php

@@ -0,0 +1,70 @@
+<?php
+
+namespace MyCLabs\Work\Worker;
+
+use PhpAmqpLib\Channel\AMQPChannel;
+
+/**
+ * RabbitMQ implementation.
+ *
+ * @author Matthieu Napoli <[email protected]>
+ */
+class RabbitMQWorker extends Worker
+{
+    /**
+     * @var AMQPChannel
+     */
+    private $channel;
+
+    /**
+     * @var string
+     */
+    private $queue;
+
+    /**
+     * @param AMQPChannel $channel
+     * @param string      $queue
+     */
+    public function __construct(AMQPChannel $channel, $queue)
+    {
+        $this->channel = $channel;
+        $this->queue = $queue;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function work()
+    {
+        $callback = function($message) {
+            $this->workHandler($message);
+        };
+
+        $this->channel->basic_qos(null, 1, null);
+        $this->channel->basic_consume($this->queue, '', false, false, false, false, $callback);
+
+        // Loop infinitely to execute tasks
+        while(count($this->channel->callbacks)) {
+            $this->channel->wait();
+        }
+    }
+
+    /**
+     * Handles a message
+     *
+     * @param mixed $message
+     */
+    private function workHandler($message)
+    {
+        /** @var AMQPChannel $channel */
+        $channel = $message->delivery_info['channel'];
+
+        $task = unserialize($message->body);
+
+        // Execute the task
+        $this->getExecutor($task)->execute($task);
+
+        // Send ACK signaling the task execution is over
+        $channel->basic_ack($message->delivery_info['delivery_tag']);
+    }
+}