Add support for all callable type not just Closure

composer.json

@@ -40,8 +40,12 @@
     ],
     "autoload": {
         "psr-0": {
-            "Nats": "src",
-            "Nats\\Test": "src"
+            "Nats": "src"
+        }
+    },
+    "autoload-dev": {
+        "psr-4": {
+            "Nats\\tests\\Unit\\": "test/"
         }
     }
 }

phpunit.xml.dist

@@ -12,7 +12,7 @@
 
     <testsuites>
         <testsuite name="PHPNats Test Suite">
-            <directory suffix=".php">./tests/Unit/</directory>
+            <directory suffix=".php">./test/</directory>
         </testsuite>
     </testsuites>
     <logging>

src/Nats/Connection.php

@@ -408,7 +408,7 @@ private function handleMSG($line)
         if (count($parts) === 5) {
             $length  = trim($parts[4]);
             $subject = $parts[3];
-        } else if (count($parts) === 4) {
+        } elseif (count($parts) === 4) {
             $length  = trim($parts[3]);
             $subject = $parts[1];
         }
@@ -481,11 +481,11 @@ public function ping()
      *
      * @param string   $subject  Message topic.
      * @param string   $payload  Message data.
-     * @param \Closure $callback Closure to be executed as callback.
+     * @param callable $callback Closure to be executed as callback.
      *
      * @return void
      */
-    public function request($subject, $payload, \Closure $callback)
+    public function request($subject, $payload, callable $callback)
     {
         $inbox = uniqid('_INBOX.');
         $sid   = $this->subscribe(
@@ -501,11 +501,11 @@ public function request($subject, $payload, \Closure $callback)
      * Subscribes to an specific event given a subject.
      *
      * @param string   $subject  Message topic.
-     * @param \Closure $callback Closure to be executed as callback.
+     * @param callable $callback Closure to be executed as callback.
      *
      * @return string
      */
-    public function subscribe($subject, \Closure $callback)
+    public function subscribe($subject, callable $callback)
     {
         $sid = $this->randomGenerator->generateString(16);
         $msg = 'SUB '.$subject.' '.$sid;
@@ -519,11 +519,11 @@ public function subscribe($subject, \Closure $callback)
      *
      * @param string   $subject  Message topic.
      * @param string   $queue    Queue name.
-     * @param \Closure $callback Closure to be executed as callback.
+     * @param callable $callback Closure to be executed as callback.
      *
      * @return string
      */
-    public function queueSubscribe($subject, $queue, \Closure $callback)
+    public function queueSubscribe($subject, $queue, callable $callback)
     {
         $sid = $this->randomGenerator->generateString(16);
         $msg = 'SUB '.$subject.' '.$queue.' '.$sid;
@@ -587,7 +587,7 @@ public function wait($quantity = 0)
     {
         $count = 0;
         $info  = stream_get_meta_data($this->streamSocket);
-        while (is_resource($this->streamSocket) === true && feof($this->streamSocket) === false && empty($info['timed_out']) === true) {
+        while (is_resource($this->streamSocket) && feof($this->streamSocket) === false && empty($info['timed_out'])) {
             $line = $this->receive();
 
             if ($line === false) {

src/Nats/EncodedConnection.php

@@ -36,11 +36,11 @@ public function __construct(ConnectionOptions $options = null, Encoder $encoder
      *
      * @param string   $subject  Message topic.
      * @param string   $payload  Message data.
-     * @param \Closure $callback Closure to be executed as callback.
+     * @param callable $callback Closure to be executed as callback.
      *
      * @return void
      */
-    public function request($subject, $payload, \Closure $callback)
+    public function request($subject, $payload, callable $callback)
     {
         $payload = $this->encoder->encode($payload);
         parent::request($subject, $payload, $callback);
@@ -65,11 +65,11 @@ public function publish($subject, $payload = null, $inbox = null)
      * Subscribes to an specific event given a subject.
      *
      * @param string   $subject  Message topic.
-     * @param \Closure $callback Closure to be executed as callback.
+     * @param callable $callback Closure to be executed as callback.
      *
      * @return string
      */
-    public function subscribe($subject, \Closure $callback)
+    public function subscribe($subject, callable $callback)
     {
         $c = function ($message) use ($callback) {
             $message->setBody($this->encoder->decode($message->getBody()));
@@ -83,11 +83,11 @@ public function subscribe($subject, \Closure $callback)
      *
      * @param string   $subject  Message topic.
      * @param string   $queue    Queue name.
-     * @param \Closure $callback Closure to be executed as callback.
+     * @param callable $callback Closure to be executed as callback.
      *
      * @return void
      */
-    public function queueSubscribe($subject, $queue, \Closure $callback)
+    public function queueSubscribe($subject, $queue, callable $callback)
     {
         $c = function ($message) use ($callback) {
             $message->setBody($this->encoder->decode($message->getBody()));

test/CallableClass.php

@@ -0,0 +1,40 @@
+<?php
+namespace Nats\tests\Unit;
+
+class CallableClass
+{
+    private $tmpStorage = [];
+    private $tmpCallbacks = [];
+
+    public function requestSubTest($res)
+    {
+        $res->reply('Hello, '.$res->getBody());
+    }
+    public function __invoke($res)
+    {
+        $msg = $res->getBody();
+        $p = explode(' ', $msg);
+        $count = array_pop($p);
+        if (isset($this->tmpCallbacks[$count])) {
+            $cb = $this->tmpCallbacks[$count];
+            $cb($msg, $count);
+            unset($this->tmpCallbacks[$count]);
+        } else {
+            $this->tmpStorage[$count] = $msg;
+        }
+    }
+
+    public function getMsg()
+    {
+        return $this->msg;
+    }
+
+    public function test($i, callable $param)
+    {
+        if (isset($this->tmpStorage[$i])) {
+            $param($this->tmpStorage[$i], $i);
+        } else {
+            $this->tmpCallbacks[$i] = $param;
+        }
+    }
+}

test/ConnectionOptionsTest.php

@@ -18,7 +18,16 @@ class ConnectionOptionsTest extends \PHPUnit_Framework_TestCase
     public function testSettersAndGetters()
     {
         $options = new ConnectionOptions();
-        $options->setHost('host')->setPort(4222)->setUser('user')->setPass('password')->setLang('lang')->setVersion('version')->setVerbose(true)->setPedantic(true)->setReconnect(true);
+        $options
+            ->setHost('host')
+            ->setPort(4222)
+            ->setUser('user')
+            ->setPass('password')
+            ->setLang('lang')
+            ->setVersion('version')
+            ->setVerbose(true)
+            ->setPedantic(true)
+            ->setReconnect(true);
 
         $this->assertEquals('host', $options->getHost());
         $this->assertEquals(4222, $options->getPort());
@@ -40,7 +49,15 @@ public function testSettersAndGetters()
     public function testAuthToken()
     {
         $options = new ConnectionOptions();
-        $options->setHost('host')->setPort(4222)->setToken('token')->setLang('lang')->setVersion('version')->setVerbose(true)->setPedantic(true)->setReconnect(true);
+        $options
+            ->setHost('host')
+            ->setPort(4222)
+            ->setToken('token')
+            ->setLang('lang')
+            ->setVersion('version')
+            ->setVerbose(true)
+            ->setPedantic(true)
+            ->setReconnect(true);
 
         $this->assertEquals('host', $options->getHost());
         $this->assertEquals(4222, $options->getPort());
@@ -63,7 +80,14 @@ public function testAuthToken()
     public function testSettersAndGettersWithoutCredentials()
     {
         $options = new ConnectionOptions();
-        $options->setHost('host')->setPort(4222)->setLang('lang')->setVersion('version')->setVerbose(true)->setPedantic(true)->setReconnect(true);
+        $options
+            ->setHost('host')
+            ->setPort(4222)
+            ->setLang('lang')
+            ->setVersion('version')
+            ->setVerbose(true)
+            ->setPedantic(true)
+            ->setReconnect(true);
 
         $this->assertEquals('host', $options->getHost());
         $this->assertEquals(4222, $options->getPort());

test/ConnectionTest.php

@@ -225,4 +225,33 @@ public function testRefusedConnection()
         $c->connect();
         $this->assertFalse($this->c->isConnected());
     }
+
+    /**
+     * Test having a class a callable
+     *
+     * @return void
+     */
+    public function testCallableRequest()
+    {
+        $testCallable = new CallableClass();
+
+        $i = 0;
+        do {
+            $this->c->subscribe(
+                'sayhello' . $i,
+                [$testCallable, 'requestSubTest']
+            );
+
+            /* Test __invoke */
+            $this->c->request(
+                'sayhello' . $i,
+                'McFly ' . $i,
+                $testCallable
+            );
+            $testCallable->test($i, function ($msg, $i) {
+                $this->assertEquals('Hello, McFly ' . $i, $msg);
+            });
+            $i++;
+        } while ($i < 100);
+    }
 }