Fixed request <-> reply bug and correctly getting the response

Author: repejota

src/Nats/Connection.php

@@ -3,6 +3,7 @@
 
 use RandomLib\Factory;
 use RandomLib\Generator;
+use Symfony\Component\Validator\Constraints\True;
 
 /**
  * Connection Class.
@@ -15,18 +16,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 +53,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 +340,10 @@ private function send($payload)
                 break;
             }
         }
+
+        if ($this->debug === true) {
+            printf('>>>> %s', $msg);
+        }
     }
 
     /**
@@ -348,6 +373,10 @@ private function receive($len = 0)
             $line = fgets($this->streamSocket);
         }
 
+        if ($this->debug === true) {
+            printf('<<<< %s\r\n', $line);
+        }
+
         return $line;
     }
 
@@ -460,15 +489,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);
     }
 
@@ -510,30 +536,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

@@ -51,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
         );
     }