Merge branch 'release/0.4.0'

Author: repejota

VERSION

@@ -1 +1 @@
-0.3.0
+0.4.0

src/Connection.php

@@ -137,9 +137,15 @@ private function send($payload)
     private function receive($len = null)
     {
         if ($len) {
-            return trim(fgets($this->streamSocket, $len + 1));
+            $line = fgets($this->streamSocket, $len + 1);
         } else {
-            return trim(fgets($this->streamSocket));
+            $line = fgets($this->streamSocket);
+        }
+
+        if ($line === false) {
+            return $line;
+        } else {
+            return trim($line);
         }
     }
 
@@ -242,12 +248,12 @@ public function publish($subject, $payload)
     /**
      * Subscribes to an specific event given a subject.
      *
-     * @param string $subject  Message topic.
-     * @param mixed  $callback Closure to be executed as callback.
+     * @param string   $subject  Message topic.
+     * @param \Closure $callback Closure to be executed as callback.
      *
      * @return string
      */
-    public function subscribe($subject, $callback)
+    public function subscribe($subject, \Closure $callback)
     {
         $sid = uniqid();
         $msg = 'SUB '.$subject.' '.$sid;
@@ -327,13 +333,14 @@ public function wait($quantity = 0)
         $count = 0;
         while (!feof($this->streamSocket)) {
             $line = $this->receive();
+            if ($line === false) {
+                return null;
+            }
 
-            // PING
             if (strpos($line, 'PING') === 0) {
                 $this->handlePING();
             }
 
-            // MSG
             if (strpos($line, 'MSG') === 0) {
                 $count = $count + 1;
                 $this->handleMSG($line);
@@ -347,6 +354,26 @@ public function wait($quantity = 0)
         return $this;
     }
 
+    /**
+     * Set Stream Timeout.
+     *
+     * @param integer $seconds Before timeout on stream.
+     *
+     * @return boolean
+     */
+    public function setStreamTimeout($seconds)
+    {
+        if ($this->isConnected()) {
+            try {
+                return stream_set_timeout($this->streamSocket, $seconds);
+            } catch (\Exception $e) {
+                return false;
+            }
+        }
+
+        return false;
+    }
+
     /**
      * Reconnects to the server.
      *

tests/Unit/ConnectionTest.php

@@ -11,22 +11,28 @@
 class ConnectionTest extends \PHPUnit_Framework_TestCase
 {
     /**
+     * Client.
+     *
      * @var resource Client
      */
     private $c;
 
     /**
-     * @var resource A separated process
+     * Process.
+     *
+     * @var resource A separated process.
      */
     private static $process;
 
     /**
+     * Gnatsd switch.
+     *
      * @var bool Am I using a real or a fake server?
      */
     private static $isGnatsd = false;
 
     /**
-     * Before Class code setup
+     * Before Class code setup.
      *
      * @return void
      */
@@ -41,7 +47,7 @@ public static function setUpBeforeClass()
     }
 
     /**
-     * After Class code setup
+     * After Class code setup.
      *
      * @return void
      */
@@ -53,7 +59,7 @@ public static function tearDownAfterClass()
     }
 
     /**
-     * setUp test suite
+     * SetUp test suite.
      *
      * @return void
      */
@@ -155,19 +161,56 @@ public function testSubscription()
     }
 
     /**
-     * Test Request command
+     * Test Request command.
      *
      * @return void
      */
     public function testRequest()
     {
-        $this->c->subscribe("sayhello", function ($res) {
-            $res->reply("Hello, ".$res->getBody(). " !!!");
-        });
+        $this->c->subscribe(
+            "sayhello",
+            function ($res) {
+                $res->reply("Hello, ".$res->getBody(). " !!!");
+            }
+        );
+
+        $this->c->request(
+            'sayhello',
+            'McFly',
+            function ($message) {
+                $this->assertNotNull($message);
+                $this->assertEquals($message, 'Hello, McFly !!!');
+            }
+        );
+    }
 
-        $this->c->request('sayhello', 'McFly', function ($message) {
-            $this->assertNotNull($message);
-            $this->assertEquals($message, 'Hello, McFly !!!');
-        });
+    /**
+     * Test Unsubscribe command.
+     *
+     * @return void
+     */
+    public function testUnsubscribe()
+    {
+        $sid = $this->c->subscribe(
+            "unsub",
+            function ($res) {
+                $this->assertTrue(false);
+            }
+        );
+        $this->c->unsubscribe($sid);
+        $this->c->publish('unsub', 'bar');
+
+        $this->assertTrue(true);
+    }
+
+    /**
+     * Test setStreamTimeout command.
+     *
+     * @return void
+     */
+    public function testSetStreamTimeout()
+    {
+        $this->assertTrue($this->c->setStreamTimeout(2));
+        $this->assertFalse($this->c->setStreamTimeout("hello"));
     }
 }