Merge pull request #4 from mbonneau/rfc_dep

Remove RFC code depend on RFC project

Author: mbonneau

.travis.yml

@@ -11,12 +11,12 @@ matrix:
 
 before_install:
     - export PATH=$HOME/.local/bin:$PATH
-    - pip install autobahntestsuite --user `whoami`
-    - pip list autobahntestsuite --user `whoami`
+    - pip install --user autobahntestsuite
+    - pip list --user autobahntestsuite
 
 before_script:
     - composer install
     - sh test/ab/run_ab_tests.sh
 
 script:
-    - phpunit
+    - vendor/bin/phpunit

CHANGELOG.md

@@ -0,0 +1,7 @@
+# 0.10.0
+
+## Changes/Additions
+
+- Project now uses [RFC6455](https://github.com/ratchetphp/RFC6455) library for underlying protocol support
+- Message subject now emits `Ratchet\RFC6455\Messaging\Message` instead of `Rx\Websocket\Message`
+- `Client` is no longer a `Subject`

README.md

@@ -6,21 +6,19 @@ Rx\Websocket is a PHP Websocket library.
 
 #### Client
 ```php
-<?php
+$client = new \Rx\Websocket\Client('ws://127.0.0.1:9191/');
 
-require_once __DIR__ . '/vendor/autoload.php';
-
-$client = new \Rx\Websocket\Client("ws://127.0.0.1:9191/");
-
-$client->subscribe(new \Rx\Observer\CallbackObserver(
+$client->subscribeCallback(
     function (\Rx\Websocket\MessageSubject $ms) {
-        $ms->subscribe(new \Rx\Observer\CallbackObserver(
+        $ms->subscribeCallback(
             function ($message) {
                 echo $message . "\n";
             }
-        ));
+        );
 
-        $sayHello = function () use ($ms) { $ms->onNext("Hello"); };
+        $sayHello = function () use ($ms) {
+            $ms->onNext('Hello');
+        };
 
         $sayHello();
         \EventLoop\addPeriodicTimer(5, $sayHello);
@@ -31,43 +29,27 @@ $client->subscribe(new \Rx\Observer\CallbackObserver(
     function () {
         // stopped trying to connect here
     }
-));
+);
 ```
 
 #### An Echo Server
 ```php
-<?php
-
-require_once __DIR__ . "/vendor/autoload.php";
-
-$server = new \Rx\Websocket\Server("127.0.0.1", 9191);
+$server = new \Rx\Websocket\Server('127.0.0.1', 9191);
 
-$server
-    ->subscribe(new \Rx\Observer\CallbackObserver(
-        function (\Rx\Websocket\MessageSubject $cs) {
-            $cs->subscribe($cs);
-        }
-    ));
+$server->subscribeCallback(function (\Rx\Websocket\MessageSubject $cs) {
+    $cs->subscribe($cs);
+});
 ```
 
 #### Server that dumps everything to the console
 ```php
-<?php
-
-require_once __DIR__ . '/vendor/autoload.php';
-
 $server = new \Rx\Websocket\Server('127.0.0.1', 9191);
 
-$server
-    ->subscribe(new \Rx\Observer\CallbackObserver(
-        function (\Rx\Websocket\MessageSubject $cs) {
-            $cs->subscribe(new \Rx\Observer\CallbackObserver(
-                function ($message) {
-                    echo $message;
-                }
-            ));
-        }
-    ));
+$server->subscribeCallback(function (\Rx\Websocket\MessageSubject $cs) {
+    $cs->subscribeCallback(function ($message) {
+        echo $message;
+    });
+});
 ```
 
 ## Installation

composer.json

@@ -22,16 +22,19 @@
             "name": "David Dan", "email": "[email protected]", "role": "Developer"
         }
     ],
-    "require": {
-        "react/http": "^0.4.1",
-        "react/http-client": "^0.4.8",
-        "voryx/event-loop": "^0.2.0",
-        "reactivex/rxphp": "^1.0",
-        "guzzlehttp/psr7": "^1.2"
-    },
     "autoload": {
         "psr-4": {
-            "Rx\\Websocket\\": "src/Websocket/"
+            "Rx\\Websocket\\": "src/"
         }
+    },
+    "require": {
+        "react/http": "^0.4.1",
+        "react/http-client": "^0.4.10",
+        "voryx/event-loop": "^0.2.1",
+        "reactivex/rxphp": "^1.5.1",
+        "ratchet/rfc6455": "^0.2.2"
+    },
+    "require-dev": {
+        "phpunit/phpunit": "^5.7.0"
     }
 }

phpunit.xml

@@ -13,9 +13,6 @@
     <testsuites>
         <testsuite name="RxWebsocket Test">
             <directory>test/</directory>
-            <exclude>
-                <directory>test/ab</directory>
-            </exclude>
         </testsuite>
     </testsuites>
 

src/Websocket/Client.php src/Client.php

@@ -2,18 +2,19 @@
 
 namespace Rx\Websocket;
 
-use Exception;
-use Rx\Websocket\RFC6455\Handshake\ClientNegotiator;
+use GuzzleHttp\Psr7\Uri;
+use Ratchet\RFC6455\Handshake\ClientNegotiator;
 use React\Dns\Resolver\Factory;
 use React\HttpClient\Request;
 use React\HttpClient\Response;
 use Rx\Disposable\CallbackDisposable;
+use Rx\Observable;
 use Rx\Observable\AnonymousObservable;
 use Rx\Observer\CallbackObserver;
 use Rx\ObserverInterface;
-use Rx\Subject\Subject;
+use Rx\SchedulerInterface;
 
-class Client extends Subject
+class Client extends Observable
 {
     /** @var string */
     protected $url;
@@ -32,40 +33,46 @@ class Client extends Subject
      */
     public function __construct($url, $useMessageObject = false, array $subProtocols = [])
     {
-        $this->url = $url;
+        $this->url              = $url;
         $this->useMessageObject = $useMessageObject;
-        $this->subProtocols = $subProtocols;
+        $this->subProtocols     = $subProtocols;
     }
 
-    private function startConnection()
+    public function subscribe(ObserverInterface $clientObserver, $scheduler = null)
     {
         $loop = \EventLoop\getLoop();
 
         $dnsResolverFactory = new Factory();
+
         $dnsResolver = $dnsResolverFactory->createCached('8.8.8.8', $loop);
 
         $factory = new \React\HttpClient\Factory();
-        $client = $factory->create($loop, $dnsResolver);
+        $client  = $factory->create($loop, $dnsResolver);
+
+        $cNegotiator = new ClientNegotiator();
 
-        $cNegotiator = new ClientNegotiator($this->url, $this->subProtocols);
+        /** @var \GuzzleHttp\Psr7\Request $nRequest */
+        $nRequest = $cNegotiator->generateRequest(new Uri($this->url));
+
+        if (!empty($this->subProtocols)) {
+            $nRequest = $nRequest
+                ->withoutHeader('Sec-WebSocket-Protocol')
+                ->withHeader('Sec-WebSocket-Protocol', $this->subProtocols);
+        }
 
-        $headers = $cNegotiator->getRequest()->getHeaders();
+        $headers = $nRequest->getHeaders();
 
         $flatHeaders = [];
         foreach ($headers as $k => $v) {
             $flatHeaders[$k] = $v[0];
         }
 
-        $request = $client->request("GET", $this->url, $flatHeaders, '1.1');
+        $request = $client->request('GET', $this->url, $flatHeaders, '1.1');
 
-        $request->on('response', function (Response $response, Request $request) use ($cNegotiator) {
+        $request->on('response', function (Response $response, Request $request) use ($flatHeaders, $cNegotiator, $nRequest, $clientObserver) {
             if ($response->getCode() !== 101) {
-                throw new \Exception("Unexpected response code " . $response->getCode());
+                throw new \Exception('Unexpected response code ' . $response->getCode());
             }
-            // TODO: Should validate response
-            //$cNegotiator->validateResponse($response);
-
-            $subprotoHeader = "";
 
             $psr7Response = new \GuzzleHttp\Psr7\Response(
                 $response->getCode(),
@@ -74,12 +81,17 @@ private function startConnection()
                 $response->getVersion()
             );
 
-            if (count($psr7Response->getHeader('Sec-WebSocket-Protocol')) == 1) {
-                $subprotoHeader = $psr7Response->getHeader('Sec-WebSocket-Protocol')[0];
+            $psr7Request = new \GuzzleHttp\Psr7\Request('GET', $this->url, $flatHeaders);
+
+            if (!$cNegotiator->validateResponse($psr7Request, $psr7Response)) {
+                throw new \Exception('Invalid response');
             }
 
-            parent::onNext(new MessageSubject(
-                new AnonymousObservable(function (ObserverInterface $observer) use ($response) {
+            $subprotoHeader = $psr7Response->getHeader('Sec-WebSocket-Protocol');
+
+            $clientObserver->onNext(new MessageSubject(
+                new AnonymousObservable(function (ObserverInterface $observer, SchedulerInterface $scheduler) use ($response, $clientObserver) {
+
                     $response->on('data', function ($data) use ($observer) {
                         $observer->onNext($data);
                     });
@@ -92,18 +104,19 @@ private function startConnection()
                         $observer->onCompleted();
                     });
 
-                    $response->on('end', function () use ($observer) {
+                    $response->on('end', function () use ($observer, $clientObserver) {
                         $observer->onCompleted();
 
                         // complete the parent observer - we only do 1 connection
-                        parent::onCompleted();
+                        $clientObserver->onCompleted();
                     });
 
 
                     return new CallbackDisposable(function () use ($response) {
                         // commented this out because disposal was causing the other
                         // end (the request) to close also - which causes the pending messages
                         // to get tossed
+                        // maybe try with $response->end()?
                         //$response->close();
                     });
                 }),
@@ -121,41 +134,15 @@ function () use ($request) {
                 true,
                 $this->useMessageObject,
                 $subprotoHeader,
-                $cNegotiator->getRequest(),
+                $nRequest,
                 $psr7Response
             ));
         });
 
         $request->writeHead();
-    }
-
-    public function subscribe(ObserverInterface $observer, $scheduler = null)
-    {
-        if (!$this->isStopped) {
-            $this->startConnection();
-        }
-
-        return parent::subscribe($observer, $scheduler);
-    }
-
-    public function send($value)
-    {
-        $this->onNext($value);
-    }
-
-    // Not sure we need this object to be a subject - just being an observer should be good enough I think
-    public function onNext($value)
-    {
-
-    }
-
-    public function onError(Exception $exception)
-    {
-
-    }
-
-    public function onCompleted()
-    {
 
+        return new CallbackDisposable(function () use ($request) {
+            $request->close();
+        });
     }
 }