Merge branch 'release/0.8.6'

Author: repejota

VERSION

@@ -1 +1 @@
-0.8.5
+0.8.6

examples/connect.php

@@ -1,13 +1,6 @@
 <?php
 require_once __DIR__.'/../vendor/autoload.php';
 
-$connectionOptions = new \Nats\ConnectionOptions(
-    [
-     'host' => '127.0.0.1',
-     'port' => 4222,
-    ]
-);
-
-$c = new Nats\Connection($connectionOptions);
+$c = new Nats\Connection();
 $c->connect();
 $c->close();

examples/ping.php

@@ -9,4 +9,6 @@
 $c = new Nats\Connection($connectionOptions);
 $c->connect();
 
+
+// TODO ping is not public.
 $c->ping();

examples/pubsub/pub.php

@@ -0,0 +1,9 @@
+<?php
+require_once __DIR__.'/../../vendor/autoload.php';
+
+$client = new \Nats\Connection();
+$client->connect();
+
+// Simple Publisher.
+$client->publish('foo', 'bar');
+$client->close();

examples/pubsub/pubsub.php

@@ -18,3 +18,5 @@ function ($message) {
 
 // Wait for 1 message.
 $client->wait(1);
+
+$client->close();

examples/pubsub/sub.php

@@ -0,0 +1,15 @@
+<?php
+require_once __DIR__.'/../../vendor/autoload.php';
+
+$client = new \Nats\Connection();
+$client->connect();
+
+// Simple Subscriber.
+$client->subscribe(
+    'foo',
+    function ($message) {
+        printf("Data: %s\r\n", $message->getBody());
+    }
+);
+
+$client->close();

examples/reqres/request.php

@@ -0,0 +1,18 @@
+<?php
+require_once __DIR__.'/../../vendor/autoload.php';
+
+use Nats\Connection;
+
+$client = new Connection();
+$client->connect();
+
+// Request.
+$client->request(
+    'foo',
+    'Marty McFly',
+    function ($message) {
+        echo $message->getBody();
+    }
+);
+
+$client->close();

src/Nats/Connection.php

@@ -15,18 +15,31 @@ class Connection
 {
 
     /**
-     * Number of PINGs.
+     * Show DEBUG info?
      *
-     * @var integer number of pings.
+     * @var boolean $debug If debug is enabled.
      */
-    private $pings = 0;
+    private $debug = false;
+
 
     /**
-     * Chunk size in bytes to use when reading an stream of data.
+     * Enable or disable debug mode.
      *
-     * @var integer size of chunk.
+     * @param boolean $debug If debug is enabled.
+     *
+     * @return void
      */
-    private $chunkSize = 1500;
+    public function setDebug($debug)
+    {
+        $this->debug = $debug;
+    }
+
+    /**
+     * Number of PINGs.
+     *
+     * @var integer number of pings.
+     */
+    private $pings = 0;
 
 
     /**
@@ -39,6 +52,13 @@ public function pingsCount()
         return $this->pings;
     }
 
+    /**
+     * Chunk size in bytes to use when reading an stream of data.
+     *
+     * @var integer size of chunk.
+     */
+    private $chunkSize = 1500;
+
     /**
      * Number of messages published.
      *
@@ -319,6 +339,10 @@ private function send($payload)
                 break;
             }
         }
+
+        if ($this->debug === true) {
+            printf('>>>> %s', $msg);
+        }
     }
 
     /**
@@ -348,6 +372,10 @@ private function receive($len = 0)
             $line = fgets($this->streamSocket);
         }
 
+        if ($this->debug === true) {
+            printf('<<<< %s\r\n', $line);
+        }
+
         return $line;
     }
 
@@ -460,14 +488,12 @@ public function ping()
     public function request($subject, $payload, \Closure $callback)
     {
         $inbox = uniqid('_INBOX.');
-        $this->subscribe(
+        $sid   = $this->subscribe(
             $inbox,
             $callback
         );
-        $msg = 'PUB '.$subject.' '.$inbox.' '.strlen($payload);
-        $this->send($msg."\r\n".$payload);
-        $this->pubs += 1;
-
+        $this->unsubscribe($sid, 1);
+        $this->publish($subject, $payload, $inbox);
         $this->wait(1);
     }
 
@@ -509,30 +535,43 @@ public function queueSubscribe($subject, $queue, \Closure $callback)
     /**
      * Unsubscribe from a event given a subject.
      *
-     * @param string $sid Subscription ID.
+     * @param string  $sid      Subscription ID.
+     * @param integer $quantity Quantity of messages.
      *
      * @return void
      */
-    public function unsubscribe($sid)
+    public function unsubscribe($sid, $quantity = null)
     {
         $msg = 'UNSUB '.$sid;
+        if ($quantity !== null) {
+            $msg = $msg.' '.$quantity;
+        }
+
         $this->send($msg);
-        unset($this->subscriptions[$sid]);
+        if ($quantity === null) {
+            unset($this->subscriptions[$sid]);
+        }
     }
 
     /**
      * Publish publishes the data argument to the given subject.
      *
      * @param string $subject Message topic.
      * @param string $payload Message data.
+     * @param string $inbox   Message inbox.
      *
      * @return void
-
+     *
      * @throws Exception If subscription not found.
      */
-    public function publish($subject, $payload = null)
+    public function publish($subject, $payload = null, $inbox = null)
     {
-        $msg = 'PUB '.$subject.' '.strlen($payload);
+        $msg = 'PUB '.$subject;
+        if ($inbox !== null) {
+            $msg = $msg.' '.$inbox;
+        }
+
+        $msg = $msg.' '.strlen($payload);
         $this->send($msg."\r\n".$payload);
         $this->pubs += 1;
     }

src/Nats/EncodedConnection.php

@@ -1,6 +1,8 @@
 <?php
 namespace Nats;
 
+use Nats\Encoders\Encoder;
+
 /**
  * Class EncodedConnection
  *
@@ -23,7 +25,7 @@ class EncodedConnection extends Connection
      * @param ConnectionOptions           $options Connection options object.
      * @param \Nats\Encoders\Encoder|null $encoder Encoder to use with the payload.
      */
-    public function __construct(ConnectionOptions $options = null, \Nats\Encoders\Encoder $encoder = null)
+    public function __construct(ConnectionOptions $options = null, Encoder $encoder = null)
     {
         $this->encoder = $encoder;
         parent::__construct($options);
@@ -49,13 +51,14 @@ public function request($subject, $payload, \Closure $callback)
      *
      * @param string $subject Message topic.
      * @param string $payload Message data.
+     * @param string $inbox   Message inbox.
      *
      * @return void
      */
-    public function publish($subject, $payload = null)
+    public function publish($subject, $payload = null, $inbox = null)
     {
         $payload = $this->encoder->encode($payload);
-        parent::publish($subject, $payload);
+        parent::publish($subject, $payload, $inbox);
     }
 
     /**

src/Nats/Message.php

@@ -176,8 +176,8 @@ public function getConn()
      */
     public function reply($body)
     {
-        $this->getConn()->publish(
-            $this->getSubject(),
+        $this->conn->publish(
+            $this->subject,
             $body
         );
     }