Merge pull request repejota#45 from dfeyer/queuesupport

TASK: Implement Connection::QueueSubscribe

Author: repejota

src/Connection.php

@@ -296,6 +296,25 @@ public function subscribe($subject, \Closure $callback)
         return $sid;
     }
 
+    /**
+     * Subscribes to an specific event given a subject and a queue.
+     *
+     * @param string   $subject  Message topic.
+     * @param string   $queue    Queue name.
+     * @param \Closure $callback Closure to be executed as callback.
+     *
+     * @return string
+     */
+    public function queueSubscribe($subject, $queue, \Closure $callback)
+    {
+        $sid = uniqid();
+        $msg = 'SUB '.$subject.' '.$queue.' '. $sid;
+        $this->send($msg);
+        $this->subscriptions[$sid] = $callback;
+
+        return $sid;
+    }
+
     /**
      * Unsubscribe from a event given a subject.
      *

tests/Unit/ConnectionTest.php

@@ -157,6 +157,33 @@ public function testSubscription()
         $this->c->wait(1);
     }
 
+    /**
+     * Test Queue Subscription command.
+     *
+     * @return void
+     */
+    public function testQueueSubscription()
+    {
+        $callback = function ($message) {
+            $this->assertNotNull($message);
+            $this->assertEquals($message, 'bar');
+        };
+
+        $this->c->queueSubscribe('foo', 'bar', $callback);
+        $this->assertGreaterThan(0, $this->c->subscriptionsCount());
+        $subscriptions = $this->c->getSubscriptions();
+        $this->assertInternalType('array', $subscriptions);
+
+        $this->c->publish('foo', 'bar');
+        $this->assertEquals(1, $this->c->pubsCount());
+/*
+        $process = new BackgroundProcess('/usr/bin/php ./tests/Util/ClientServerStub.php ');
+        $process->run();
+*/
+        // time_nanosleep(1, 0);
+        $this->c->wait(1);
+    }
+
     /**
      * Test Request command.
      *