Merge pull request repejota#97 from repejota/feature/encoded_connections

WIP: Feature/encoded connections

Author: repejota

.travis.yml

@@ -3,7 +3,6 @@ services:
   - docker
 language: php
 php:
-  - 5.5
   - 5.6
   - 7
   - 7.1

README.md

@@ -21,7 +21,7 @@ A PHP client for the [NATS messaging system](https://nats.io).
 Requirements
 ------------
 
-* php 5.5+
+* php 5.6+
 * [nats](https://github.com/derekcollison/nats) or [gnatsd](https://github.com/apcera/gnatsd)
 
 

composer.json

@@ -8,7 +8,7 @@
         "ircmaxell/random-lib": "^1.1"
     },
     "require-dev": {
-        "phpunit/phpunit": "4.7.*",
+        "phpunit/phpunit": "5.*",
         "satooshi/php-coveralls": "dev-master",
         "squizlabs/php_codesniffer": "~2.0",
         "phpspec/phpspec": "~2.0"

examples/pubsub/jsonencodedpub.php

@@ -0,0 +1,17 @@
+<?php
+require_once __DIR__.'/../../vendor/autoload.php';
+
+$encoder           = new \Nats\Encoders\JSONEncoder();
+$connectionOptions = new \Nats\ConnectionOptions();
+
+$connectionOptions->setHost('localhost')->setPort(4222);
+$c = new Nats\EncodedConnection($connectionOptions, $encoder);
+$c->connect();
+
+$c->reconnect();
+
+$c->publish('foo', 'bar');
+$c->publish('foo', 'bar');
+$c->publish('foo', 'bar');
+$c->publish('foo', 'bar');
+$c->publish('foo', 'bar');

examples/pubsub/jsonencodedsub.php

@@ -0,0 +1,19 @@
+<?php
+require_once __DIR__.'/../../vendor/autoload.php';
+
+$encoder           = new \Nats\Encoders\JSONEncoder();
+$connectionOptions = new \Nats\ConnectionOptions();
+
+$connectionOptions->setHost('localhost')->setPort(4222);
+$c = new Nats\EncodedConnection($connectionOptions, $encoder);
+$c->connect();
+
+$callback = function ($payload) {
+    printf("Data: %s\r\n", $payload);
+};
+
+$sid = $c->subscribe('foo', $callback);
+
+$c->wait(2);
+
+$c->unsubscribe($sid);

spec/Nats/ConnectionOptionsSpec.php

@@ -33,7 +33,7 @@ function it_has_default_lang_value_as_php() {
     }
 
     function it_has_default_version_value_as_null() {
-        $this->getVersion()->shouldEqual("0.8.0");
+        $this->getVersion()->shouldEqual("0.8.2");
     }
 
     function it_has_default_verbose_value_as_null() {

src/Nats/Connection.php

@@ -6,21 +6,23 @@
 
 /**
  * Connection Class.
+ *
+ * Handles the connection to a NATS server or cluster of servers.
  */
 class Connection
 {
 
     /**
-     * Number of PINGS.
+     * Number of PINGs.
      *
-     * @var int number of pings
+     * @var integer number of pings.
      */
     private $pings = 0;
 
     /**
-     * Chunk size in bytes to use when reading with fread.
+     * Chunk size in bytes to use when reading an stream of data.
      *
-     * @var integer
+     * @var integer size of chunk.
      */
     private $chunkSize = 1500;
 
@@ -313,23 +315,24 @@ public function ping()
     /**
      * 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.
+     * @param string $subject Message topic.
+     * @param string $payload Message data.
      *
      * @return void
      */
-    public function request($subject, $payload, $callback, $wait = 1)
+    public function request($subject, $payload)
     {
         $inbox = uniqid('_INBOX.');
-        $this->subscribe($inbox, $callback);
-
+        $this->subscribe(
+            $inbox,
+            function ($message) {
+            }
+        );
         $msg = 'PUB '.$subject.' '.$inbox.' '.strlen($payload);
         $this->send($msg."\r\n".$payload);
         $this->pubs += 1;
 
-        $this->wait($wait);
+        $this->wait(1);
     }
 
 
@@ -363,7 +366,6 @@ public function subscribe($subject, \Closure $callback)
         $msg = 'SUB '.$subject.' '.$sid;
         $this->send($msg);
         $this->subscriptions[$sid] = $callback;
-
         return $sid;
     }
 
@@ -383,7 +385,6 @@ public function queueSubscribe($subject, $queue, \Closure $callback)
         $msg = 'SUB '.$subject.' '.$queue.' '.$sid;
         $this->send($msg);
         $this->subscriptions[$sid] = $callback;
-
         return $sid;
     }
 
@@ -399,7 +400,6 @@ public function unsubscribe($sid)
     {
         $msg = 'UNSUB '.$sid;
         $this->send($msg);
-
         unset($this->subscriptions[$sid]);
     }
 
@@ -420,9 +420,8 @@ private function handlePING()
      *
      * @param string $line Message command from Nats.
      *
-     * @return             void
-     * @throws             Exception If subscription not found.
-     * @codeCoverageIgnore
+     * @return void
+     * @throws Exception If subscription not found.
      */
     private function handleMSG($line)
     {
@@ -452,8 +451,6 @@ private function handleMSG($line)
         } else {
             throw Exception::forSubscriptionCallbackInvalid($sid);
         }
-
-        return;
     }
 
 

src/Nats/EncodedConnection.php

@@ -0,0 +1,94 @@
+<?php
+namespace Nats;
+
+/**
+ * Class EncodedConnection
+ *
+ * @package Nats
+ */
+class EncodedConnection extends Connection
+{
+
+    /**
+     * Encoder for this connection.
+     *
+     * @var \Nats\Encoders\Encoder|null
+     */
+    private $encoder = null;
+
+
+    /**
+     * EncodedConnection constructor.
+     *
+     * @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)
+    {
+        $this->encoder = $encoder;
+        parent::__construct($options);
+    }
+
+    /**
+     * Request does a request and executes a callback with the response.
+     *
+     * @param string $subject Message topic.
+     * @param string $payload Message data.
+     *
+     * @return void
+     */
+    public function request($subject, $payload)
+    {
+        $payload = $this->encoder->encode($payload);
+        parent::request($subject, $payload);
+    }
+
+    /**
+     * 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 = null)
+    {
+        $payload = $this->encoder->encode($payload);
+        parent::publish($subject, $payload);
+    }
+
+    /**
+     * Subscribes to an specific event given a subject.
+     *
+     * @param string   $subject  Message topic.
+     * @param \Closure $callback Closure to be executed as callback.
+     *
+     * @return void
+     */
+    public function subscribe($subject, \Closure $callback)
+    {
+        $c = function ($message) use ($callback) {
+            $message->setBody($this->encoder->decode($message->getBody()));
+            $callback($message);
+        };
+        parent::subscribe($subject, $c);
+    }
+
+    /**
+     * Subscribes to an specific event given a subject and a queue.
+     *
+     * @param string   $subject  Message topic.
+     * @param string   $queue    Queue name.
+     * @param \Closure $callback Closure to be executed as callback.
+     *
+     * @return void
+     */
+    public function queueSubscribe($subject, $queue, \Closure $callback)
+    {
+        $c = function ($message) use ($callback) {
+            $message->setBody($this->encoder->decode($message->getBody()));
+            $callback($message);
+        };
+        parent::queueSubscribe($subject, $queue, $c);
+    }
+}

src/Nats/Encoders/Encoder.php

@@ -0,0 +1,30 @@
+<?php
+namespace Nats\Encoders;
+
+/**
+ * Interface Encoder
+ *
+ * @package Nats\Encoders
+ */
+interface Encoder
+{
+
+
+    /**
+     * Encodes a message.
+     *
+     * @param string $payload Message to decode.
+     *
+     * @return mixed
+     */
+    public function encode($payload);
+
+    /**
+     * Decodes a message.
+     *
+     * @param string $payload Message to decode.
+     *
+     * @return mixed
+     */
+    public function decode($payload);
+}

src/Nats/Encoders/JSONEncoder.php

@@ -0,0 +1,40 @@
+<?php
+namespace Nats\Encoders;
+
+/**
+ * Class JSONEncoder
+ *
+ * Encodes and decodes messages in JSON format.
+ *
+ * @package Nats
+ */
+class JSONEncoder implements Encoder
+{
+
+
+    /**
+     * Encodes a message to JSON.
+     *
+     * @param string $payload Message to decode.
+     *
+     * @return mixed
+     */
+    public function encode($payload)
+    {
+        $payload = json_encode($payload);
+        return $payload;
+    }
+
+    /**
+     * Decodes a message from JSON.
+     *
+     * @param string $payload Message to decode.
+     *
+     * @return mixed
+     */
+    public function decode($payload)
+    {
+        $payload = json_decode($payload);
+        return $payload;
+    }
+}