Merge branch 'release/0.8.0'

Author: repejota

Makefile

@@ -9,11 +9,13 @@ cs: lint
 	./vendor/bin/phpcs --standard=PSR2 --warning-severity=0 src test examples
 	./vendor/bin/phpcs --standard=Squiz --sniffs=Squiz.Commenting.FunctionComment,Squiz.Commenting.FunctionCommentThrowTag,Squiz.Commenting.ClassComment,Squiz.Commenting.VariableComment src test examples
 
-test:
+test: tdd bdd
+
+tdd:
 	./vendor/bin/phpunit test
 
 bdd:
-	./vendor/bin/phpspec run --format=pretty
+	./vendor/bin/phpspec run --format=pretty -v
 
 cover:
 	./vendor/bin/phpunit --coverage-html ./cover test

VERSION

@@ -1 +1 @@
-0.7.2
+0.8.0

spec/Nats/ConnectionOptionsSpec.php

@@ -33,7 +33,7 @@ function it_has_default_lang_value_as_php() {
     }
 
     function it_has_default_version_value_as_null() {
-        $this->getVersion()->shouldEqual("0.0.5");
+        $this->getVersion()->shouldEqual("0.8.0");
     }
 
     function it_has_default_verbose_value_as_null() {

spec/Nats/ConnectionSpec.php

@@ -41,7 +41,7 @@ function it_is_disconnected()
         $this->isConnected()->shouldBe(false);
     }
 
-    function it_can_connect_and_disconnect_with_default_options()
+    function it_connects_and_disconnects_with_default_options()
     {
         $this->connect();
         $this->shouldHaveType('Nats\Connection');
@@ -59,7 +59,7 @@ function it_increases_reconnects_count_on_each_reconnection()
         $this->close();
     }
 
-    function it_a_ping_is_sent_after_a_successful_connection()
+    function it_sends_ping_after_a_successful_connection()
     {
         $this->connect();
         $this->pingsCount()->shouldBe(1);
@@ -94,5 +94,30 @@ function it_decreases_subscriptions_after_unsubscribing_to_a_topic()
         $this->close();
     }
 
+    function it_sends_a_message_with_a_1024c_subject()
+    {
+        $this->connect();
+        $subject = str_pad("", 1024*1, "x");
+        $this->publish($subject);
+        $this->pubsCount()->shouldBe(1);
+        $this->close();
+    }
+
+    function it_sends_a_message_with_a_1024x10c_subject()
+    {
+        $this->connect();
+        $subject = str_pad("", 1024*10, "x");
+        $this->publish($subject);
+        $this->pubsCount()->shouldBe(1);
+        $this->close();
+    }
 
+    function it_sends_a_message_with_a_1024x100c_subject()
+    {
+        $this->connect();
+        $subject = str_pad("", 1024*100, "x");
+        $this->publish($subject);
+        $this->pubsCount()->shouldBe(1);
+        $this->close();
+    }
 }

src/Nats/Connection.php

@@ -190,14 +190,11 @@ private function receive($len = null)
      * @return resource
      * @throws \Exception Exception raised if connection fails.
      */
-    private function getStream($address, $timeout = null)
+    private function getStream($address, $timeout)
     {
-        if (is_null($timeout)) {
-            $timeout = intval(ini_get('default_socket_timeout'));
-        }
         $errno = null;
         $errstr = null;
-        
+
         $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT);
         $timeout = number_format($timeout, 3);
         $seconds = floor($timeout);
@@ -231,9 +228,13 @@ public function isConnected()
      */
     public function connect($timeout = null)
     {
+        if ($timeout === null) {
+            $timeout = intval(ini_get('default_socket_timeout'));
+        }
 
         $this->timeout = $timeout;
         $this->streamSocket = $this->getStream($this->options->getAddress(), $timeout);
+        $this->setStreamTimeout($timeout);
 
         $msg = 'CONNECT '.$this->options;
         $this->send($msg);
@@ -413,7 +414,8 @@ private function handleMSG($line)
     public function wait($quantity = 0)
     {
         $count = 0;
-        while (!feof($this->streamSocket)) {
+        $info = stream_get_meta_data($this->streamSocket);
+        while (is_resource($this->streamSocket) && !feof($this->streamSocket) && !$info['timed_out']) {
             $line = $this->receive();
 
             if ($line === false) {
@@ -431,6 +433,7 @@ public function wait($quantity = 0)
                     return $this;
                 }
             }
+            $info = stream_get_meta_data($this->streamSocket);
         }
         $this->close();
 
@@ -447,9 +450,12 @@ public function wait($quantity = 0)
     public function setStreamTimeout($seconds)
     {
         if ($this->isConnected()) {
-            if (is_int($seconds)) {
+            if (is_numeric($seconds)) {
                 try {
-                    return $this->streamWrapper->setStreamTimeout($this->streamSocket, $seconds);
+                    $timeout = number_format($seconds, 3);
+                    $seconds = floor($timeout);
+                    $microseconds = ($timeout - $seconds) * 1000;
+                    return stream_set_timeout($this->streamSocket, $seconds, $microseconds);
                 } catch (\Exception $e) {
                     return false;
                 }
@@ -489,4 +495,13 @@ public function close()
         fclose($this->streamSocket);
         $this->streamSocket = null;
     }
+
+
+    /**
+     * @return resource
+     */
+    public function streamSocket()
+    {
+        return $this->streamSocket;
+    }
 }

src/Nats/ConnectionOptions.php

@@ -47,7 +47,7 @@ class ConnectionOptions
      *
      * @var string
      */
-    private $version = "0.0.5";
+    private $version = "0.8.0";
 
     /**
      * If verbose mode is enabled.

test/ConnectionOptionsTest.php

@@ -74,7 +74,7 @@ public function testSettersAndGettersWithoutCredentials()
     public function testStringRepresentation()
     {
         $options = new ConnectionOptions();
-        $this->assertEquals("{\"lang\":\"php\",\"version\":\"0.0.5\",\"verbose\":false,\"pedantic\":false}", $options->__toString());
+        $this->assertEquals("{\"lang\":\"php\",\"version\":\"0.8.0\",\"verbose\":false,\"pedantic\":false}", $options->__toString());
     }
 
     /**
@@ -87,6 +87,6 @@ public function testStringRepresentationWithCredentials()
         $options = new ConnectionOptions();
         $options->setUser("username");
         $options->setPass("password");
-        $this->assertEquals("{\"lang\":\"php\",\"version\":\"0.0.5\",\"verbose\":false,\"pedantic\":false,\"user\":\"username\",\"pass\":\"password\"}", $options->__toString());
+        $this->assertEquals("{\"lang\":\"php\",\"version\":\"0.8.0\",\"verbose\":false,\"pedantic\":false,\"user\":\"username\",\"pass\":\"password\"}", $options->__toString());
     }
 }

test/ConnectionTest.php

@@ -128,8 +128,8 @@ function ($message) {
     public function testLargeRequest()
     {
 
-        $content = file_get_contents(dirname(__FILE__).'/test.pdf');
-        
+        $content = substr(str_shuffle("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"), 0, 51200);
+
         $contentLen = strlen($content);
 
         $contentSum = md5($content);
@@ -161,6 +161,32 @@ function ($message) use ($contentLen) {
 
     }
 
+    /**
+     * Test setting a timeout on the stream
+     *
+     * @return void
+     */
+    public function testSetStreamTimeout()
+    {
+        $this->c->setStreamTimeout(1);
+        $before = time();
+        $this->c->request(
+            "nonexistantsubject",
+            "test",
+            function ($message) {
+                $this->fail("should never have gotten here");
+            }
+        );
+        $timeTaken = time() - $before;
+
+        $this->assertGreaterThan(0, $timeTaken);
+        $this->assertLessThan(3, $timeTaken);
+
+        $meta = stream_get_meta_data($this->c->streamSocket());
+
+        $this->assertTrue($meta["timed_out"]);
+    }
+
     /**
      * Test Unsubscribe command.
      *