Merge branch 'master' into develop

Author: repejota

Makefile

@@ -5,9 +5,9 @@ lint:
 	find examples -name *.php -exec php -l {} \;
 
 cs: lint
-	./bin/phpcbf --standard=PSR2 src tests examples
-	./bin/phpcs --standard=PSR2 --warning-severity=0 src tests examples
-	./bin/phpcs --standard=Squiz --sniffs=Squiz.Commenting.FunctionComment,Squiz.Commenting.FunctionCommentThrowTag,Squiz.Commenting.ClassComment,Squiz.Commenting.VariableComment src tests examples
+	./vendor/bin/phpcbf --standard=PSR2 src test examples
+	./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: tdd bdd
 

src/Nats/Connection.php

@@ -18,9 +18,9 @@ class Connection
 
     /**
      * Chunk size in bytes to use when reading with fread.
-     * @var int
+     * @var integer
      */
-    private $chunkSize = 8192;
+    private $chunkSize = 1500;
 
     /**
      * Return the number of pings.
@@ -167,12 +167,13 @@ private function receive($len = null)
             $receivedBytes = 0;
             while ($receivedBytes < $len) {
                 $bytesLeft = $len - $receivedBytes;
-                if ( $bytesLeft < $this->chunkSize ) {
+                if ($bytesLeft < $this->chunkSize) {
                     $chunkSize = $bytesLeft;
                 }
 
-                $line .= fread($this->streamSocket, $chunkSize);
-                $receivedBytes += $chunkSize;
+                $readChunk = fread($this->streamSocket, $chunkSize);
+                $receivedBytes += strlen($readChunk);
+                $line .= $readChunk;
             }
         } else {
             $line = fgets($this->streamSocket);
@@ -183,8 +184,8 @@ private function receive($len = null)
     /**
      * Returns an stream socket to the desired server.
      *
-     * @param string  $address Server url string.
-     * @param float $timeout Number of seconds until the connect() system call should timeout.
+     * @param string $address Server url string.
+     * @param float  $timeout Number of seconds until the connect() system call should timeout.
      *
      * @return resource
      * @throws \Exception Exception raised if connection fails.
@@ -196,7 +197,7 @@ private function getStream($address, $timeout = null)
         }
         $errno = null;
         $errstr = null;
-        
+
         $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT);
         $timeout = number_format($timeout, 3);
         $seconds = floor($timeout);
@@ -367,7 +368,7 @@ private function handlePING()
      * @param string $line Message command from Nats.
      *
      * @return void
-     * @throws Exception
+     * @throws Exception If subscription not found.
      * @codeCoverageIgnore
      */
     private function handleMSG($line)
@@ -470,9 +471,11 @@ public function reconnect()
     }
 
     /**
-     * @param integer $chunkSize Set byte chunk len to read when reading from wire
+     * @param integer $chunkSize Set byte chunk len to read when reading from wire.
+     * @return void
      */
-    public function setChunkSize($chunkSize){
+    public function setChunkSize($chunkSize)
+    {
         $this->chunkSize = $chunkSize;
     }
 

test/ConnectionTest.php

@@ -13,7 +13,7 @@ class ConnectionTest extends \PHPUnit_Framework_TestCase
     /**
      * Client.
      *
-     * @var resource Client
+     * @var Nats\Connection Client
      */
     private $c;
 
@@ -97,21 +97,68 @@ public function testReconnect()
      */
     public function testRequest()
     {
-        $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 !!!');
-            }
-        );
+        $i = 0;
+        do {
+            $this->c->subscribe(
+                "sayhello$i",
+                function ($res) {
+                    $res->reply("Hello, ".$res->getBody(). " !!!");
+                }
+            );
+
+            $this->c->request(
+                "sayhello$i",
+                'McFly',
+                function ($message) {
+                    $this->assertNotNull($message);
+                    $this->assertEquals($message, 'Hello, McFly !!!');
+                }
+            );
+
+            $i++;
+        } while ($i < 100);
+    }
+
+    /**
+     * Test Request command with large payload.
+     *
+     * @return void
+     */
+    public function testLargeRequest()
+    {
+
+        $content = file_get_contents(dirname(__FILE__).'/test.pdf');
+        
+        $contentLen = strlen($content);
+
+        $contentSum = md5($content);
+
+        $i = 0;
+        do {
+            $this->c->subscribe(
+                "saybighello$i",
+                function ($res) use ($contentLen, $contentSum) {
+                    $gotLen = strlen($res->getBody());
+                    $gotSum = md5($res->getBody());
+                    $this->assertEquals($contentLen, $gotLen);
+                    $this->assertEquals($contentSum, $gotSum);
+                    $res->reply($gotLen);
+                }
+            );
+
+            $this->c->request(
+                "saybighello$i",
+                $content,
+                function ($message) use ($contentLen) {
+                    $this->assertNotNull($message);
+                    $this->assertEquals($message->getBody(), $contentLen);
+                }
+            );
+
+            $i++;
+        } while ($i < 100);
+
     }
 
     /**