Merge branch 'release/0.7.1'

Author: repejota

AUTHORS

@@ -1,4 +1 @@
 Raül Pérez <repejota@gmail.com> - http://repejota.com - @repejota
-Adrià Cidre <adria.cidre@gmail.com> - http://oridoki.com - @adriacidre
-José Gil <josgilmo@gmail.com> - @josgilmo
-Gorka López de Torre <glopezdetorre@gmail.com> - http://gorka.io - @glopezdetorre

CONTRIBUTORS

@@ -0,0 +1,10 @@
+Adrià Cidre <adria.cidre@gmail.com> - http://oridoki.com - @adriacidre
+José Gil <josgilmo@gmail.com> - @josgilmo
+Gorka López de Torre <glopezdetorre@gmail.com> - http://gorka.io - @glopezdetorre
+Issel Guberna - http://isselguberna.com/ - @octante
+Donal Byrne - <byrnedo@tcd.ie> - http://www.byrnedo.com/
+Dominique Feyer - <dfeyer@ttree.ch> - http://www.ttree.ch/
+Oliver Mack -
+Superbug - http://www.superbug.co/
+netroby - http://www.netroby.com/
+

README.md

@@ -113,23 +113,6 @@ Creators
 - <https://twitter.com/repejota>
 - <https://github.com/repejota>
 
-**Adrià Cidre**
-
-- <https://twitter.com/adriacidre>
-- <https://github.com/adriacidre>
-
-**José Gil**
-
-- <https://twitter.com/josgilmo>
-- <https://github.com/josgilmo>
-
-**Gorka López de Torre**
-
-- <https://twitter.com/glopezdetorre>
-- <https://github.com/glopezdetorre>
-
-
-
 License
 -------
 

VERSION

@@ -1 +1 @@
-0.7.0
+0.7.1

composer.json

@@ -5,6 +5,7 @@
     "minimum-stability": "dev",
     "license": "MIT",
     "require": {
+        "ircmaxell/random-lib": "^1.1"
     },
     "require-dev": {
         "phpunit/phpunit": "4.7.*",

examples/connectauth.php

@@ -0,0 +1,12 @@
+<?php
+require_once __DIR__ . "/../vendor/autoload.php";
+
+$connectionOptions = new \Nats\ConnectionOptions();
+$connectionOptions
+    ->setHost('localhost')
+    ->setPort(4222)
+    ->setUser("foo")
+    ->setPass("bar");
+$c = new Nats\Connection($connectionOptions);
+$c->connect();
+$c->close();

src/Connection.php

@@ -1,6 +1,9 @@
 <?php
 namespace Nats;
 
+use RandomLib\Factory;
+use RandomLib\Generator;
+
 /**
  * Connection Class.
  */
@@ -13,6 +16,12 @@ class Connection
      */
     private $pings = 0;
 
+    /**
+     * Chunk size in bytes to use when reading with fread.
+     * @var int
+     */
+    private $chunkSize = 8192;
+
     /**
      * Return the number of pings.
      *
@@ -91,6 +100,13 @@ public function getSubscriptions()
      */
     private $options = null;
 
+    /**
+     * Connection timeout
+     *
+     * @var float
+     */
+    private $timeout = null;
+
     /**
      * Stream File Pointer.
      *
@@ -105,6 +121,11 @@ public function getSubscriptions()
      */
     private $streamWrapper;
 
+    /**
+     * @var Generator
+     */
+    private $randomGenerator;
+
     /**
      * Constructor.
      *
@@ -117,6 +138,8 @@ public function __construct(ConnectionOptions $options = null)
         $this->subscriptions = [];
         $this->options = $options;
         $this->streamWrapper = new StreamWrapper();
+        $randomFactory = new Factory();
+        $this->randomGenerator = $randomFactory->getLowStrengthGenerator();
 
         if (is_null($options)) {
             $this->options = new ConnectionOptions();
@@ -159,7 +182,21 @@ private function receive($len = null)
     {
 
         if ($len) {
-            $line = fread($this->streamSocket, $len);
+            $chunkSize = $this->chunkSize;
+            $line = null;
+            $receivedBytes = 0;
+            while ($receivedBytes < $len) {
+                $bytesLeft = $len - $receivedBytes;
+                if ( $bytesLeft < 1500 ) {
+                    $chunkSize = $bytesLeft;
+                }
+
+                $line .= fread($this->streamSocket, $chunkSize);
+                $receivedBytes += $chunkSize;
+            }
+            if (strlen($line) > 2) {
+                $line = substr($line, 0, -2);
+            }
         } else {
             $line = fgets($this->streamSocket);
         }
@@ -170,7 +207,7 @@ private function receive($len = null)
      * Returns an stream socket to the desired server.
      *
      * @param string  $address Server url string.
-     * @param integer $timeout Number of seconds until the connect() system call should timeout.
+     * @param float $timeout Number of seconds until the connect() system call should timeout.
      *
      * @return resource
      * @throws \Exception Exception raised if connection fails.
@@ -206,13 +243,14 @@ public function isConnected()
     /**
      * Connect to server.
      *
-     * @param integer $timeout Number of seconds until the connect() system call should timeout.
+     * @param float $timeout Number of seconds until the connect() system call should timeout.
      *
      * @throws \Exception Exception raised if connection fails.
      * @return void
      */
     public function connect($timeout = null)
     {
+        $this->timeout = $timeout;
         $this->streamSocket = $this->getStream($this->options->getAddress(), $timeout);
         $msg = 'CONNECT '.$this->options;
         $this->send($msg);
@@ -271,7 +309,7 @@ public function request($subject, $payload, $callback, $wait = 1)
      *
      * @return void
      */
-    public function publish($subject, $payload)
+    public function publish($subject, $payload = null)
     {
         $msg = 'PUB '.$subject.' '.strlen($payload);
         $this->send($msg . "\r\n" . $payload);
@@ -288,7 +326,7 @@ public function publish($subject, $payload)
      */
     public function subscribe($subject, \Closure $callback)
     {
-        $sid = uniqid();
+        $sid = $this->randomGenerator->generateString(16);
         $msg = 'SUB '.$subject.' '.$sid;
         $this->send($msg);
         $this->subscriptions[$sid] = $callback;
@@ -307,7 +345,7 @@ public function subscribe($subject, \Closure $callback)
      */
     public function queueSubscribe($subject, $queue, \Closure $callback)
     {
-        $sid = uniqid();
+        $sid = $this->randomGenerator->generateString(16);
         $msg = 'SUB '.$subject.' '.$queue.' '. $sid;
         $this->send($msg);
         $this->subscriptions[$sid] = $callback;
@@ -345,7 +383,8 @@ private function handlePING()
      *
      * @param string $line Message command from NATS.
      *
-     * @return \Exception|void
+     * @return void
+     * @throws Exception
      * @codeCoverageIgnore
      */
     private function handleMSG($line)
@@ -366,11 +405,15 @@ private function handleMSG($line)
         $payload = $this->receive($length);
         $msg = new Message($subject, $payload, $sid, $this);
 
+        if (!isset($this->subscriptions[$sid])) {
+            throw new Exception('subscription not found');
+        }
+
         $func = $this->subscriptions[$sid];
         if (is_callable($func)) {
             $func($msg);
         } else {
-            return new \Exception('not callable');
+            throw new Exception('not callable');
         }
 
         return;
@@ -413,7 +456,7 @@ public function wait($quantity = 0)
     /**
      * Set Stream Timeout.
      *
-     * @param integer $seconds Before timeout on stream.
+     * @param float $seconds Before timeout on stream.
      *
      * @return boolean
      */
@@ -441,7 +484,14 @@ public function reconnect()
     {
         $this->reconnects += 1;
         $this->close();
-        $this->connect();
+        $this->connect($this->timeout);
+    }
+
+    /**
+     * @param integer $chunkSize Set byte chunk len to read when reading from wire
+     */
+    public function setChunkSize($chunkSize){
+        $this->chunkSize = $chunkSize;
     }
 
     /**

src/Exception.php

@@ -0,0 +1,10 @@
+<?php
+namespace Nats;
+
+/**
+ * Exception Class.
+ */
+class Exception extends \Exception
+{
+
+}

src/Message.php

@@ -30,7 +30,7 @@ class Message
     /**
      * Message related connection.
      *
-     * @var string
+     * @var Connection
      */
     private $conn;
 
@@ -146,7 +146,7 @@ public function setConn(Connection $conn)
     /**
      * Get Conn.
      *
-     * @return string
+     * @return Connection
      */
     public function getConn()
     {

src/StreamWrapper.php

@@ -10,30 +10,35 @@ class StreamWrapper
     /**
      * Wrapper for stream_socket_client
      *
-     * @param string  $address    Address to connect the socket.
-     * @param integer $errno      Number of error.
-     * @param string  $errstr     Description of error.
-     * @param integer $timeout    Timeout.
+     * @param string $address Address to connect the socket.
+     * @param integer $errno Number of error.
+     * @param string $errstr Description of error.
+     * @param float $timeout Timeout.
      * @param integer $typeStream Type of stream.
      *
-     * @return stream
+     * @return resource
      */
     public function getStreamSocketClient($address, &$errno, &$errstr, $timeout, $typeStream)
     {
-        return stream_socket_client($address, $errno, $errstr, $timeout, $typeStream);
+        $stream = stream_socket_client($address, $errno, $errstr, $timeout, $typeStream);
+        $this->setStreamTimeout($stream, $timeout);
+        return $stream;
     }
-    
+
     /**
      * Wrapper for stream_set_timeout
      *
-     * @param mixed   $stream  Stream.
-     * @param integer $seconds Seconds for timeout.
+     * @param mixed $stream Stream.
+     * @param float $seconds Seconds for timeout.
      *
      * @return boolean
      *
-    */
+     */
     public function setStreamTimeout($stream, $seconds)
     {
-        return stream_set_timeout($stream, $seconds);
+        $timeout = number_format($seconds, 3);
+        $seconds = floor($timeout);
+        $microseconds = ($timeout - $seconds) * 1000;
+        return stream_set_timeout($stream, $seconds, $microseconds);
     }
 }