Initial commit

Author: mbonneau

.gitignore

@@ -0,0 +1,3 @@
+/vendor
+/composer.lock
+/test/ab/reports

README.md

@@ -0,0 +1,80 @@
+RxWebsocket is a PHP Websocket library.
+
+## Usage
+
+#### Client
+```php
+<?php
+
+require_once __DIR__ . '/vendor/autoload.php';
+
+$client = new \RxWebsocket\Client("ws://127.0.0.1:9191/");
+
+$client->subscribe(new \Rx\Observer\CallbackObserver(
+    function (\RxWebsocket\MessageSubject $ms) {
+        $ms->subscribe(new \Rx\Observer\CallbackObserver(
+            function ($message) {
+                echo $message . "\n";
+            }
+        ));
+
+        $sayHello = function () use ($ms) { $ms->onNext("Hello"); };
+
+        $sayHello();
+        \EventLoop\addPeriodicTimer(5, $sayHello);
+    },
+    function ($error) {
+        // connection errors here
+    },
+    function () {
+        // stopped trying to connect here
+    }
+));
+```
+
+#### An Echo Server
+```php
+<?php
+
+require_once __DIR__ . "/vendor/autoload.php";
+
+$server = new \RxWebsocket\Server("127.0.0.1", 9191);
+
+$server
+    ->subscribe(new \Rx\Observer\CallbackObserver(
+        function (\RxWebsocket\MessageSubject $cs) {
+            $cs->subscribe($cs);
+        }
+    ));
+```
+
+#### Server that dumps everything to the console
+```php
+<?php
+
+require_once __DIR__ . "/vendor/autoload.php";
+
+$server = new \RxWebsocket\Server("127.0.0.1", 9191);
+
+$server
+    ->subscribe(new \Rx\Observer\CallbackObserver(
+        function (\RxWebsocket\MessageSubject $cs) {
+            $ms->subscribe(new CallbackObserver(
+                function ($message) {
+                    echo $message;
+                }
+            ));
+        }
+    ));
+```
+
+## Installation
+
+Using [composer](https://getcomposer.org/):
+
+Right now, this project uses components of the [Ratchet RFC6455 project](https://github.com/ratchetphp/RFC6455) that
+are not tagged with a release, so they must be installed manually before this project:
+```composer require ratchet/rfc6455:dev-psr```
+
+Then install this project:
+```composer require voryx/rxwebsocket```

composer.json

@@ -0,0 +1,37 @@
+{
+    "name": "voryx/rxwebsocket",
+    "type": "library",
+    "description": "Async Websockets for PHP using Rx",
+    "keywords": [
+        "websocket",
+        "websockets",
+        "rfc6455",
+        "rx",
+        "rx.php",
+        "rxphp",
+        "react",
+        "reactive"
+    ],
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Matt Bonneau", "email": "[email protected]", "role": "Developer"
+        },
+
+        {
+            "name": "David Dan", "email": "[email protected]", "role": "Developer"
+        }
+    ],
+    "require": {
+        "react/http": "dev-master",
+        "react/http-client": "dev-master",
+        "ratchet/rfc6455": "dev-psr7",
+        "asm89/rx.php": "dev-master",
+        "voryx/event-loop": "^0.2.0"
+    },
+    "autoload": {
+        "psr-4": {
+            "Voryx\\RxWebsocket\\": "src/RxWebsocket/"
+        }
+    }
+}

src/RxWebsocket/Client.php

@@ -0,0 +1,133 @@
+<?php
+
+namespace Voryx\RxWebsocket;
+
+use Exception;
+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\AnonymousObservable;
+use Rx\Observer\CallbackObserver;
+use Rx\ObserverInterface;
+use Rx\Subject\Subject;
+
+class Client extends Subject
+{
+    protected $url;
+
+    /**
+     * RxWebsocket constructor.
+     * @param $url
+     */
+    public function __construct($url)
+    {
+        $this->url = $url;
+    }
+
+    private function startConnection()
+    {
+        $loop = \EventLoop\getLoop();
+
+        $dnsResolverFactory = new Factory();
+        $dnsResolver = $dnsResolverFactory->createCached('8.8.8.8', $loop);
+
+        $factory = new \React\HttpClient\Factory();
+        $client = $factory->create($loop, $dnsResolver);
+
+        $cNegotiator = new ClientNegotiator($this->url);
+
+        $headers = $cNegotiator->getRequest()->getHeaders();
+
+        $flatHeaders = [];
+        foreach ($headers as $k => $v) {
+            $flatHeaders[$k] = $v[0];
+        }
+
+        $request = $client->request("GET", $this->url, $flatHeaders, '1.1');
+
+        $request->on('response', function (Response $response, Request $request) use ($cNegotiator) {
+            if ($response->getCode() !== 101) {
+                throw new \Exception("Unexpected response code " . $response->getCode());
+            }
+            // TODO: Should validate response
+            //$cNegotiator->validateResponse($response);
+
+            parent::onNext(new MessageSubject(
+                new AnonymousObservable(function (ObserverInterface $observer) use ($response) {
+                    $response->on('data', function ($data) use ($observer) {
+                        $observer->onNext($data);
+                    });
+
+                    $response->on('error', function ($e) use ($observer) {
+                        $observer->onError($e);
+                    });
+
+                    $response->on('close', function () use ($observer) {
+                        $observer->onCompleted();
+                    });
+
+                    $response->on('end', function () use ($observer) {
+                        $observer->onCompleted();
+
+                        // complete the parent observer - we only do 1 connection
+                        parent::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
+                        //$response->close();
+                    });
+                }),
+                new CallbackObserver(
+                    function ($x) use ($request) {
+                        $request->write($x);
+                    },
+                    function ($e) use ($request) {
+                        $request->close();
+                    },
+                    function () use ($request) {
+                        $request->end();
+                    }
+                ),
+                true
+            ));
+        });
+
+        $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()
+    {
+
+    }
+}