Merge branch 'release/0.8.1'

Author: repejota

.travis.yml

@@ -6,6 +6,7 @@ php:
   - 5.5
   - 5.6
   - 7
+  - 7.1
 cache:
   directories:
    - vendor

Makefile

@@ -30,4 +30,7 @@ dist-clean:
 	rm -f composer.phar
 	rm -f composer.lock
 
+docker-nats:
+	docker run --rm -p 8222:8222 -p 4222:4222 -d --name nats-main nats
+
 .PHONY: lint test cs cover deps dist-clean

README.md

@@ -19,7 +19,7 @@ A PHP client for the [NATS messaging system](https://nats.io).
 Requirements
 ------------
 
-* php 5.4+
+* php 5.5+
 * [nats](https://github.com/derekcollison/nats) or [gnatsd](https://github.com/apcera/gnatsd)
 
 

VERSION

@@ -1 +1 @@
-0.8.0
+0.8.1

src/Nats/Connection.php

@@ -115,7 +115,7 @@ public function getSubscriptions()
     private $streamSocket;
 
     /**
-     * @var Generator
+     * @var Generator|Php71RandomGenerator
      */
     private $randomGenerator;
 
@@ -130,8 +130,12 @@ public function __construct(ConnectionOptions $options = null)
         $this->pubs = 0;
         $this->subscriptions = [];
         $this->options = $options;
-        $randomFactory = new Factory();
-        $this->randomGenerator = $randomFactory->getLowStrengthGenerator();
+        if(version_compare(phpversion(), '7.0', '>')){
+            $this->randomGenerator = new Php71RandomGenerator();
+        } else {
+            $randomFactory = new Factory();
+            $this->randomGenerator = $randomFactory->getLowStrengthGenerator();
+        }
 
         if (is_null($options)) {
             $this->options = new ConnectionOptions();
@@ -148,7 +152,21 @@ public function __construct(ConnectionOptions $options = null)
     private function send($payload)
     {
         $msg = $payload."\r\n";
-        fwrite($this->streamSocket, $msg, strlen($msg));
+        $len = strlen($msg);
+        while (true) {
+            if (false === ($written = @fwrite($this->streamSocket, $msg))) {
+                throw new \Exception('Error sending data');
+            }
+            if ($written === 0) {
+                throw new \Exception('Broken pipe or closed connection');
+            }
+            $len = $len - $written;
+            if ($len > 0) {
+                $msg = substr($msg, 0 - $len);
+            } else {
+                break;
+            }
+        }
     }
 
     /**
@@ -195,16 +213,19 @@ private function getStream($address, $timeout)
         $errno = null;
         $errstr = null;
 
+        set_error_handler(function(){return true;});
         $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT);
+        restore_error_handler();
+
+        if (!$fp) {
+            throw Exception::forStreamSocketClientError($errstr, $errno);
+        }
+
         $timeout = number_format($timeout, 3);
         $seconds = floor($timeout);
         $microseconds = ($timeout - $seconds) * 1000;
         stream_set_timeout($fp, $seconds, $microseconds);
 
-        if (!$fp) {
-            throw new \Exception($errstr, $errno);
-        }
-
         return $fp;
     }
 
@@ -239,16 +260,16 @@ public function connect($timeout = null)
         $msg = 'CONNECT '.$this->options;
         $this->send($msg);
         $connect_response = $this->receive();
-        if (strpos($connect_response, '-ERR')!== false) {
-            throw new \Exception("Failing connection: $connect_response");
+
+        if ($this->isErrorResponse($connect_response)) {
+            throw Exception::forFailedConnection($connect_response);
         }
 
         $this->ping();
         $ping_response = $this->receive();
-        if ($ping_response !== "PONG") {
-            if (strpos($ping_response, '-ERR')!== false) {
-                throw new \Exception("Failing on first ping: $ping_response");
-            }
+
+        if ($this->isErrorResponse($ping_response)) {
+            throw Exception::forFailedPing($ping_response);
         }
     }
 
@@ -391,14 +412,14 @@ private function handleMSG($line)
         $msg = new Message($subject, $payload, $sid, $this);
 
         if (!isset($this->subscriptions[$sid])) {
-            throw new Exception('subscription not found');
+            throw Exception::forSubscriptionNotFound($sid);
         }
 
         $func = $this->subscriptions[$sid];
         if (is_callable($func)) {
             $func($msg);
         } else {
-            throw new Exception('not callable');
+            throw Exception::forSubscriptionCallbackInvalid($sid);
         }
 
         return;
@@ -492,6 +513,9 @@ public function setChunkSize($chunkSize)
      */
     public function close()
     {
+        if ($this->streamSocket === null) {
+            return;
+        }
         fclose($this->streamSocket);
         $this->streamSocket = null;
     }
@@ -504,4 +528,15 @@ public function streamSocket()
     {
         return $this->streamSocket;
     }
+
+    /**
+     * Indicates whether $response is an error response.
+     *
+     * @param string $response The Nats Server response.
+     * @return boolean
+     */
+    private function isErrorResponse($response)
+    {
+        return false !== strpos('-ERR', $response);
+    }
 }

src/Nats/Exception.php

@@ -6,5 +6,59 @@
  */
 class Exception extends \Exception
 {
+    /**
+     * Creates an Exception for a failed connection.
+     *
+     * @param string $response The failed error response.
+     * @return Nats\Exception
+     */
+    public static function forFailedConnection($response)
+    {
+        return new static(sprintf('Failed to connect: %s', $response));
+    }
 
+    /**
+     * Creates an Exception for a failed PING response.
+     *
+     * @param string $response The failed PING response.
+     * @return Nats\Exception
+     */
+    public static function forFailedPing($response)
+    {
+        return new static(sprintf('Failed to ping: %s', $response));
+    }
+
+    /**
+     * Creates an Exception for an invalid Subscription Identifier (sid).
+     *
+     * @param string $subscription The Subscription Identifier (sid).
+     * @return Nats\Exception
+     */
+    public static function forSubscriptionNotFound($subscription)
+    {
+        return new static(sprintf('Subscription not found: %s', $subscription));
+    }
+
+    /**
+     * Creates an Exception for an invalid Subscription Identifier (sid) callback.
+     *
+     * @param string $subscription The Subscription Identifier (sid).
+     * @return Nats\Exception
+     */
+    public static function forSubscriptionCallbackInvalid($subscription)
+    {
+        return new static(sprintf('Subscription callback is invalid: %s', $subscription));
+    }
+
+    /**
+     * Creates an Exception for the failed creation of a Stream Socket Client.
+     *
+     * @param string  $message The system level error message.
+     * @param integer $code    The system level error code.
+     * @return Nats\Exception
+     */
+    public static function forStreamSocketClientError($message, $code)
+    {
+        return new static(sprintf('A Stream Socket Client could not be created: (%d) %s', $code, $message), $code);
+    }
 }

src/Nats/Php71RandomGenerator.php

@@ -0,0 +1,19 @@
+<?php
+
+
+namespace Nats;
+
+
+
+class Php71RandomGenerator
+{
+    /**
+     * A simple wrapper on random_bytes
+     *
+     * @param $len
+     * @return string
+     */
+    public function generateString($len) {
+        return bin2hex(random_bytes($len));
+    }
+}

test/ConnectionTest.php

@@ -158,7 +158,6 @@ function ($message) use ($contentLen) {
 
             $i++;
         } while ($i < 100);
-
     }
 
     /**
@@ -205,4 +204,23 @@ function ($res) {
 
         $this->assertTrue(true);
     }
+
+
+    /**
+     * Test Connecting when nats server is not available
+     *
+     * @return void
+     */
+    public function testRefusedConnection()
+    {
+        $this->setExpectedException(\Exception::class);
+
+        $options = new ConnectionOptions();
+        $options->setHost('localhost');
+        $options->setPort(4223);
+
+        $c = new Nats\Connection($options);
+        $c->connect();
+        $this->assertFalse($this->c->isConnected());
+    }
 }