callbacks on request to handle the response

Author: repejota

src/Nats/Connection.php

@@ -421,16 +421,16 @@ public function ping()
      *
      * @param string $subject Message topic.
      * @param string $payload Message data.
+     * @param \Closure $callback Closure to be executed as callback.
      *
      * @return void
      */
-    public function request($subject, $payload)
+    public function request($subject, $payload, \Closure $callback)
     {
         $inbox = uniqid('_INBOX.');
         $this->subscribe(
             $inbox,
-            function ($message) {
-            }
+            $callback
         );
         $msg = 'PUB '.$subject.' '.$inbox.' '.strlen($payload);
         $this->send($msg."\r\n".$payload);

src/Nats/EncodedConnection.php

@@ -34,13 +34,14 @@ public function __construct(ConnectionOptions $options = null, \Nats\Encoders\En
      *
      * @param string $subject Message topic.
      * @param string $payload Message data.
+     * @param \Closure $callback Closure to be executed as callback.
      *
      * @return void
      */
-    public function request($subject, $payload)
+    public function request($subject, $payload, \Closure $callback)
     {
         $payload = $this->encoder->encode($payload);
-        parent::request($subject, $payload);
+        parent::request($subject, $payload, $callback);
     }
 
     /**
@@ -63,15 +64,15 @@ public function publish($subject, $payload = null)
      * @param string   $subject  Message topic.
      * @param \Closure $callback Closure to be executed as callback.
      *
-     * @return void
+     * @return string
      */
     public function subscribe($subject, \Closure $callback)
     {
         $c = function ($message) use ($callback) {
             $message->setBody($this->encoder->decode($message->getBody()));
             $callback($message);
         };
-        parent::subscribe($subject, $c);
+        return parent::subscribe($subject, $c);
     }
 
     /**

src/Nats/Message.php

@@ -19,7 +19,7 @@ class Message
      *
      * @var string
      */
-    private $body;
+    public $body;
 
     /**
      * Message Ssid.

test/ConnectionTest.php

@@ -113,7 +113,10 @@ function ($res) {
 
             $this->c->request(
                 'sayhello'.$i,
-                'McFly'
+                'McFly',
+                function ($res) {
+                    $this->assertEquals('Hello, McFly !!!', $res->getBody());
+                }
             );
 
             $i++;
@@ -150,9 +153,8 @@ function ($res) use ($contentLen, $contentSum) {
             $this->c->request(
                 'saybighello'.$i,
                 $content,
-                function ($message) use ($contentLen) {
-                    $this->assertNotNull($message);
-                    $this->assertEquals($message->getBody(), $contentLen);
+                function ($res) {
+
                 }
             );
 
@@ -173,8 +175,8 @@ public function testSetStreamTimeout()
         $this->c->request(
             'nonexistantsubject',
             'test',
-            function ($message) {
-                $this->fail('should never have gotten here');
+            function ($res) {
+
             }
         );
         $timeTaken = (time() - $before);

test/EncodedConnectionTest.php

@@ -82,7 +82,10 @@ function ($res) {
 
         $this->c->request(
             'sayhello',
-            'McFly'
+            'McFly',
+            function ($res) {
+                $this->assertEquals('Hello, McFly !!!', $res->getBody());
+            }
         );
     }
 
@@ -105,7 +108,11 @@ function ($res) {
             [
              'foo',
              'McFly',
-            ]
+            ],
+            function ($res) {
+                var_dump($res);
+                $this->assertEquals('Hello, McFly !!!', $res->getBody());
+            }
         );
     }
 }