Initial commit for adding TLS support

Author: tomponline

src/Nats/Connection.php

@@ -229,10 +229,10 @@ public function isConnected()
      * @param string $address Server url string.
      * @param float  $timeout Number of seconds until the connect() system call should timeout.
      *
-     * @return resource
      * @throws \Exception Exception raised if connection fails.
+     * @return resource
      */
-    private function getStream($address, $timeout)
+    private function getStream($address, $timeout, $context)
     {
         $errno  = null;
         $errstr = null;
@@ -242,7 +242,8 @@ function () {
                 return true;
             }
         );
-        $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT);
+
+        $fp = stream_socket_client($address, $errno, $errstr, $timeout, STREAM_CLIENT_CONNECT, $context);
         restore_error_handler();
 
         if ($fp === false) {
@@ -315,8 +316,8 @@ public function __construct(ConnectionOptions $options = null)
      *
      * @param string $payload Message data.
      *
-     * @return void
      * @throws \Exception Raises if fails sending data.
+     * @return void
      */
     private function send($payload)
     {
@@ -394,8 +395,8 @@ private function handlePING()
      *
      * @param string $line Message command from Nats.
      *
-     * @return             void
      * @throws             Exception If subscription not found.
+     * @return             void
      * @codeCoverageIgnore
      */
     private function handleMSG($line)
@@ -408,7 +409,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];
         }
@@ -442,20 +443,30 @@ public function connect($timeout = null)
             $timeout = intval(ini_get('default_socket_timeout'));
         }
 
+        $context = stream_context_create();
+        stream_context_set_option($context, 'ssl', 'verify_peer', false);
+        //stream_context_set_option($context, 'ssl', 'cafile', '/var/lib/puppet/ssl/certs/ca.pem');
+
         $this->timeout      = $timeout;
-        $this->streamSocket = $this->getStream($this->options->getAddress(), $timeout);
+        $this->streamSocket = $this->getStream($this->options->getAddress(), $timeout, $context);
         $this->setStreamTimeout($timeout);
 
-        $msg = 'CONNECT '.$this->options;
-        $this->send($msg);
-        $connectResponse = $this->receive();
+        $infoResponse = $this->receive();
 
-        if ($this->isErrorResponse($connectResponse) === true) {
-            throw Exception::forFailedConnection($connectResponse);
+        if ($this->isErrorResponse($infoResponse) === true) {
+            throw Exception::forFailedConnection($infoResponse);
         } else {
-            $this->processServerInfo($connectResponse);
+            $this->processServerInfo($infoResponse);
+            if ($this->serverInfo->isTLSRequired()) {
+                if (!stream_socket_enable_crypto($this->streamSocket, true,
+                                STREAM_CRYPTO_METHOD_TLSv1_2_CLIENT)) {
+                    throw Exception('Couldnt enable crypto');
+                }
+            }
         }
 
+        $msg = 'CONNECT '.$this->options;
+        $this->send($msg);
         $this->ping();
         $pingResponse = $this->receive();
 
@@ -560,9 +571,9 @@ public function unsubscribe($sid, $quantity = null)
      * @param string $payload Message data.
      * @param string $inbox   Message inbox.
      *
+     * @throws Exception If subscription not found.
      * @return void
      *
-     * @throws Exception If subscription not found.
      */
     public function publish($subject, $payload = null, $inbox = null)
     {