Merge branch 'release/0.3.0'

Author: repejota

README.md

@@ -68,6 +68,18 @@ $callback = function($payload)
 };
 $client->subscribe("foo", $callback);
 
+# Request 
+$c->request('sayhello', 'Marty McFly', function ($response) {
+    echo $response->getBody();
+});
+
+# Responding to requests
+$sid = $c->subscribe("sayhello", function ($res) {
+    $res->reply("Hello, " . $res->getBody() . " !!!");
+});
+
+
+
 # Wait for 1 message
 $client->wait(1);
 ```

VERSION

@@ -1 +1 @@
-0.2.0
+0.3.0

examples/reqres/req.php

@@ -0,0 +1,16 @@
+<?php
+require_once __DIR__ . "/../../vendor/autoload.php";
+
+$connectionOptions = new \Nats\ConnectionOptions();
+$connectionOptions->setHost('localhost')->setPort(4222);
+
+$c = new Nats\Connection($connectionOptions);
+$c->connect();
+
+$c->request(
+    'sayhello',
+    'Marty McFly',
+    function ($response) {
+        echo $response->getBody();
+    }
+);

examples/reqres/res.php

@@ -0,0 +1,20 @@
+<?php
+require_once __DIR__ . "/../../vendor/autoload.php";
+
+$connectionOptions = new \Nats\ConnectionOptions();
+$connectionOptions
+    ->setHost('localhost')
+    ->setPort(4222);
+$c = new Nats\Connection($connectionOptions);
+$c->connect();
+
+$sid = $c->subscribe(
+    "sayhello",
+    function ($res) {
+        $res->reply("Hello, " . $res->getBody() . " !!!");
+    }
+);
+
+$c->wait(2);
+
+$c->unsubscribe($sid);

src/Connection.php

@@ -85,7 +85,7 @@ public function getSubscriptions()
     }
 
     /**
-     * Connection options object
+     * Connection options object.
      *
      * @var ConnectionOptions|null
      */
@@ -118,6 +118,7 @@ public function __construct(ConnectionOptions $options = null)
      * Sends data thought the stream.
      *
      * @param string $payload Message data.
+     *
      * @return void
      */
     private function send($payload)
@@ -147,6 +148,7 @@ private function receive($len = null)
      *
      * @param string  $address Server url string.
      * @param integer $timeout Number of seconds until the connect() system call should timeout.
+     *
      * @return resource
      * @throws \Exception Exception raised if connection fails.
      */
@@ -177,6 +179,7 @@ public function isConnected()
      * Connect to server.
      *
      * @param integer $timeout Number of seconds until the connect() system call should timeout.
+     *
      * @throws \Exception Exception raised if connection fails.
      * @return void
      */
@@ -199,11 +202,34 @@ public function ping()
         $this->pings += 1;
     }
 
+    /**
+     * Request does a request and executes a callback with the response.
+     *
+     * @param string  $subject  Message topic.
+     * @param string  $payload  Message data.
+     * @param mixed   $callback Closure to be executed as callback.
+     * @param integer $wait     Number of messages to wait for.
+     *
+     * @return void
+     */
+    public function request($subject, $payload, $callback, $wait = 1)
+    {
+        $inbox = uniqid('_INBOX.');
+        $this->subscribe($inbox, $callback);
+
+        $msg = 'PUB '.$subject.' '.$inbox.' '.strlen($payload);
+        $this->send($msg . "\r\n" . $payload);
+        $this->pubs += 1;
+
+        $this->wait($wait);
+    }
+
     /**
      * Publish publishes the data argument to the given subject.
      *
      * @param string $subject Message topic.
      * @param string $payload Message data.
+     *
      * @return void
      */
     public function publish($subject, $payload)
@@ -216,8 +242,9 @@ public function publish($subject, $payload)
     /**
      * Subscribes to an specific event given a subject.
      *
-     * @param string   $subject  Message topic.
-     * @param resource $callback Closure to be executed as callback.
+     * @param string $subject  Message topic.
+     * @param mixed  $callback Closure to be executed as callback.
+     *
      * @return string
      */
     public function subscribe($subject, $callback)
@@ -234,6 +261,7 @@ public function subscribe($subject, $callback)
      * Unsubscribe from a event given a subject.
      *
      * @param string $sid Subscription ID.
+     *
      * @return void
      */
     public function unsubscribe($sid)
@@ -260,29 +288,38 @@ private function handlePING()
      * @param string $line Message command from NATS.
      *
      * @return \Exception|void
+     * @codeCoverageIgnore
      */
     private function handleMSG($line)
     {
         $parts = explode(' ', $line);
+        $subject = null;
         $length = $parts[3];
         $sid = $parts[2];
 
+        if (count($parts) == 5) {
+            $length = $parts[5];
+            $subject = $parts[3];
+        }
+
         $payload = $this->receive($length);
+        $msg = new Message($subject, $payload, $sid, $this);
 
         $func = $this->subscriptions[$sid];
         if (is_callable($func)) {
-            $func($payload);
+            $func($msg);
         } else {
             return new \Exception('not callable');
         }
-        
+
         return;
     }
 
     /**
      * Waits for messages.
      *
      * @param integer $quantity Number of messages to wait for.
+     *
      * @return resource $connection Connection object
      */
     public function wait($quantity = 0)