Merge branch 'release/0.7.0'

Author: repejota

README.md

@@ -39,7 +39,7 @@ Now let's tell composer about our project's dependancies, in this case, PHPNats.
 ```
 {
   "require": {
-    "repejota/nats": "master"
+    "repejota/nats": "dev-master"
   }
 }
 ```

VERSION

@@ -1 +1 @@
-0.6.3
+0.7.0

src/Connection.php

@@ -157,17 +157,13 @@ private function send($payload)
      */
     private function receive($len = null)
     {
+
         if ($len) {
-            $line = fgets($this->streamSocket, $len + 1);
+            $line = fread($this->streamSocket, $len);
         } else {
             $line = fgets($this->streamSocket);
         }
-
-        if ($line === false) {
-            return $line;
-        } else {
-            return trim($line);
-        }
+        return $line;
     }
 
     /**
@@ -300,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.
      *
@@ -337,14 +352,14 @@ private function handleMSG($line)
     {
         $parts = explode(' ', $line);
         $subject = null;
-        $length = $parts[3];
+        $length = trim($parts[3]);
         $sid = $parts[2];
 
         if (count($parts) == 5) {
-            $length = $parts[4];
+            $length = trim($parts[4]);
             $subject = $parts[3];
         } elseif (count($parts) == 4) {
-            $length = $parts[3];
+            $length = trim($parts[3]);
             $subject = $parts[1];
         }
 
@@ -373,6 +388,7 @@ public function wait($quantity = 0)
         $count = 0;
         while (!feof($this->streamSocket)) {
             $line = $this->receive();
+
             if ($line === false) {
                 return null;
             }

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.
      *