Now passing Autobahn tests

Author: mbonneau

src/Client.php

@@ -9,8 +9,6 @@
 use Ratchet\RFC6455\Handshake\ClientNegotiator;
 use React\EventLoop\LoopInterface;
 use React\Http\Client\Client as HttpClient;
-use React\Http\Client\Request;
-use React\Http\Message\Response;
 use React\Socket\ConnectionInterface;
 use React\Socket\ConnectorInterface;
 use Rx\Disposable\CallbackDisposable;
@@ -113,8 +111,11 @@ public function _subscribe(ObserverInterface $clientObserver): DisposableInterfa
                         $observer->onError($e);
                     });
 
-                    $request->on('close', function () use ($observer) {
+                    $request->on('close', function () use ($observer, $clientObserver) {
                         $observer->onCompleted();
+
+                        // complete the parent observer - we only do 1 connection
+                        $clientObserver->onCompleted();
                     });
 
                     $request->on('end', function () use ($observer, $clientObserver) {

src/Server.php

@@ -2,24 +2,24 @@
 
 namespace Rx\Websocket;
 
-use GuzzleHttp\Psr7\Request;
 use Psr\Http\Message\ServerRequestInterface;
 use Ratchet\RFC6455\Handshake\RequestVerifier;
 use Ratchet\RFC6455\Handshake\ServerNegotiator;
 use React\EventLoop\LoopInterface;
 use React\Http\Message\Response;
-use React\Http\Middleware\StreamingRequestMiddleware;
-use React\Http\Server as HttpServer;
-use React\Socket\Server as SocketServer;
+use React\Http\Middleware\StreamingRequestMiddleware as StreamingRequestMiddlewareAlias;
+use React\Http\HttpServer;
+use React\Socket\SocketServer;
 use React\Stream\CompositeStream;
-use React\Stream\ReadableStreamInterface;
 use React\Stream\ThroughStream;
 use Rx\Disposable\CallbackDisposable;
+use Rx\Disposable\EmptyDisposable;
 use Rx\DisposableInterface;
 use Rx\Observable;
 use Rx\Observable\AnonymousObservable;
 use Rx\Observer\CallbackObserver;
 use Rx\ObserverInterface;
+use function RingCentral\Psr7\str;
 
 class Server extends Observable
 {
@@ -41,7 +41,7 @@ public function __construct(string $bindAddressOrPort, bool $useMessageObject =
 
     public function _subscribe(ObserverInterface $observer): DisposableInterface
     {
-        $socket = new SocketServer($this->bindAddress, $this->loop);
+        $socket = new SocketServer($this->bindAddress, [], $this->loop);
 
         $negotiator = new ServerNegotiator(new RequestVerifier());
         if (!empty($this->subProtocols)) {
@@ -50,15 +50,14 @@ public function _subscribe(ObserverInterface $observer): DisposableInterface
 
         $http = new HttpServer(
             $this->loop,
-            new StreamingRequestMiddleware(),
+            new StreamingRequestMiddlewareAlias(),
             function (ServerRequestInterface $request) use ($negotiator, $observer) {
                 // cram the remote address into the header in our own X- header so
                 // the user will have access to it
                 $request = $request->withAddedHeader('X-RxWebsocket-Remote-Address', $request->getServerParams()['REMOTE_ADDR'] ?? '');
 
                 $negotiatorResponse = $negotiator->handshake($request);
 
-                /** @var ReadableStreamInterface $requestStream */
                 $requestStream  = new ThroughStream();
                 $responseStream = new ThroughStream();
 
@@ -74,8 +73,8 @@ function (ServerRequestInterface $request) use ($negotiator, $observer) {
                 );
 
                 if ($negotiatorResponse->getStatusCode() !== 101) {
-                    $responseStream->close();
-                    return;
+                    $responseStream->end(str($negotiatorResponse));
+                    return new EmptyDisposable();
                 }
 
                 $subProtocol = "";
@@ -87,7 +86,6 @@ function (ServerRequestInterface $request) use ($negotiator, $observer) {
                     new AnonymousObservable(
                         function (ObserverInterface $observer) use ($requestStream) {
                             $requestStream->on('data', function ($data) use ($observer) {
-                                var_export($data);
                                 $observer->onNext($data);
                             });
                             $requestStream->on('error', function ($error) use ($observer) {
@@ -115,7 +113,7 @@ function ($error) use ($responseStream) {
                             $responseStream->close();
                         },
                         function () use ($responseStream) {
-                            $responseStream->close();
+                            $responseStream->end();
                         }
                     ),
                     false,

test/ab/clientRunner.php

@@ -12,7 +12,14 @@
     $reportUrl = "ws://127.0.0.1:9001/updateReports?agent=" . AGENT . "&shutdownOnComplete=true";
     $client    = new \Rx\Websocket\Client($reportUrl);
 
-    $client->subscribe();
+    $client->subscribe(
+        function () {},
+        function (\Throwable $e) {
+            // wstest is mean and hangs up on you without even sending a response
+            // this is empty so we don't see HttpClient's complaints about that
+        },
+        function () {}
+    );
 };
 
 $runIndividualTest = function ($case, $timeout = 60000) {
@@ -83,6 +90,7 @@ function () use ($case, $deferred) {
     ->subscribe(
         $runTests,
         function ($error) {
-            echo $error . "\n";
-        }
+            echo "Err: " . $error . "\n";
+        },
+        function () { printf("completed.\n"); }
     );

test/ab/fuzzingclient.json

@@ -4,7 +4,7 @@
 
    "servers": [
                 {"agent": "RxWebsocketServer/0.0.0",
-                 "url": "ws://localhost:9001",
+                 "url": "ws://host.docker.internal:9001",
                  "options": {"version": 18}}
               ],
    "cases": ["*"],

test/ab/run_ab_tests.sh

@@ -1,12 +1,28 @@
 cd test/ab
 
-wstest -m fuzzingserver -s fuzzingserver.json &
-sleep 5
-php clientRunner.php
+docker run --rm \
+      -d \
+      -v ${PWD}:/config \
+      -v ${PWD}/reports:/reports \
+      -p 9001:9001 \
+      --name rxwsfuzzingserver \
+      crossbario/autobahn-testsuite wstest -m fuzzingserver -s /config/fuzzingserver.json
+sleep 10
+php -d memory_limit=256M clientRunner.php
+
+docker stop rxwsfuzzingserver
 
-sleep 2
+
+sleep 5
 
 php testServer.php 600 &
 sleep 3
-wstest -m fuzzingclient -s fuzzingclient.json
-sleep 12
+
+docker run --rm \
+       \
+      -v ${PWD}:/config \
+      -v ${PWD}/reports:/reports \
+      -p 9001:9001 \
+      --name rxwsfuzzingclient \
+      crossbario/autobahn-testsuite wstest -m fuzzingclient -s /config/fuzzingclient.json
+sleep 12

test/ab/testServer.php

@@ -11,7 +11,7 @@
     $timerObservable = Observable::timer(1000 * $argv[1]);
 }
 
-$server = new \Rx\Websocket\Server("tcp://127.0.0.1:9001", true);
+$server = new \Rx\Websocket\Server("tcp://0.0.0.0:9001", true);
 
 $server
     ->takeUntil($timerObservable)