Merge pull request #14 from fieg/queue-consume

Added queue consume support

README.md

@@ -1,4 +1,6 @@
 Queue library
 ===========================
 
+[![Build Status](https://travis-ci.org/treehouselabs/queue.svg?branch=master)](https://travis-ci.org/treehouselabs/queue)
+
 To be documented

src/TreeHouse/Queue/Message/Provider/AmqpMessageProvider.php

@@ -51,4 +51,20 @@ public function nack(Message $message, $requeue = false)
     {
         $this->queue->nack($message->getId(), $requeue ? AMQP_REQUEUE : null);
     }
+
+    /**
+     * @inheritdoc
+     */
+    public function consume(callable $callback)
+    {
+        $this->queue->consume(function(\AMQPEnvelope $envelope) use ($callback) {
+            $id    = $envelope->getDeliveryTag();
+            $body  = $envelope->getBody();
+            $props = new MessageProperties($envelope->getHeaders());
+
+            $message = new Message($body, $props, $id);
+
+            $callback($message);
+        });
+    }
 }

src/TreeHouse/Queue/Message/Provider/MessageProviderInterface.php

@@ -11,6 +11,15 @@ interface MessageProviderInterface
      */
     public function get();
 
+    /**
+     * Blocking function to consume next message
+     *
+     * @param callable $callback called when a message is available
+     *
+     * @return void
+     */
+    public function consume(callable $callback);
+
     /**
      * @param Message $message The message to ACK
      *

tests/TreeHouse/Queue/Tests/Message/Provider/AmqpMessageProviderTest.php

@@ -45,6 +45,39 @@ public function testGet()
         $this->assertEquals($headers, $message->getProperties()->toArray());
     }
 
+    public function testConsume()
+    {
+        $id      = uniqid();
+        $body    = 'test';
+        $headers = ['foo' => 'bar'];
+
+        $envelope = $this->getMock(\AMQPEnvelope::class);
+        $envelope->expects($this->any())->method('getDeliveryTag')->will($this->returnValue($id));
+        $envelope->expects($this->any())->method('getBody')->will($this->returnValue($body));
+        $envelope->expects($this->any())->method('getHeaders')->will($this->returnValue($headers));
+
+        $callback = function($message) use ($id, $body, $headers) {
+            $this->assertInstanceOf(Message::class, $message);
+
+            $this->assertEquals($id, $message->getId());
+            $this->assertEquals($body, $message->getBody());
+            $this->assertEquals($headers, $message->getProperties()->toArray());
+        };
+
+        $this->queue
+            ->expects($this->once())
+            ->method('consume')
+            ->with($this->callback(function($callback) use ($envelope) {
+                $callback($envelope);
+
+                return true;
+            }))
+        ;
+
+        $provider = new AmqpMessageProvider($this->queue);
+        $provider->consume($callback);
+    }
+
     public function testReturnNullOnEmptyEnvelope()
     {
         $this->queue