Merge pull request #121 from xp-forge/feature/websockets

Websockets support

Author: thekid

README.md

@@ -52,7 +52,6 @@ Server models
 The four server models (*selectable via `-m <model>` on the command line*) are:
 
 * **async** (*the default since 3.0.0*): A single-threaded web server. Handlers can yield control back to the server to serve other clients during lengthy operations such as file up- and downloads.
-* **sequential**: Same as above, but blocks until one client's HTTP request handler has finished executing before serving the next request.
 * **prefork**: Much like Apache, forks a given number of children to handle HTTP requests. Requires the `pcntl` extension.
 * **develop**: As mentioned above, built ontop of the PHP development wenserver. Application code is recompiled and application setup performed from scratch on every request, errors and debug output are handled by the [development console](https://github.com/xp-forge/web/pull/35).
 

composer.json

@@ -7,8 +7,9 @@
   "keywords": ["module", "xp"],
   "require" : {
     "xp-framework/core": "^12.0 | ^11.0 | ^10.0",
-    "xp-framework/networking": "^10.1 | ^9.3",
-    "xp-forge/uri": "^3.0 | ^2.0",
+    "xp-framework/networking": "^10.1",
+    "xp-forge/uri": "^3.1",
+    "xp-forge/websockets": "^4.1",
     "php": ">=7.4.0"
   },
   "require-dev" : {

src/it/php/web/unittest/IntegrationTest.class.php

@@ -1,10 +1,13 @@
 <?php namespace web\unittest;
 
-use test\{Assert, After, Test, Values};
+use peer\ProtocolException;
+use test\{Assert, After, Expect, Test, Values};
+use util\Bytes;
+use websocket\WebSocket;
 
 #[StartServer(TestingServer::class)]
 class IntegrationTest {
-  const FORM_URLENCODED = 'application/x-www-form-urlencoded';
+  const FORM_URLENCODED= 'application/x-www-form-urlencoded';
 
   private $server;
 
@@ -13,9 +16,10 @@ public function __construct($server) {
     $this->server= $server;
   }
 
-  #[After]
-  public function shutdown() {
-    $this->server->shutdown();
+  /** @return iterable */
+  private function messages() {
+    yield ['Test', 'Echo: Test'];
+    yield [new Bytes([8, 15]), new Bytes([47, 11, 8, 15])];
   }
 
   /**
@@ -161,4 +165,34 @@ public function with_large_cookie($length) {
     $r= $this->send('GET', '/cookie', '1.0', ['Cookie' => $header]);
     Assert::equals((string)strlen($header), $r['body']);
   }
+
+  #[Test, Values(from: 'messages')]
+  public function websocket_message($input, $output) {
+    try {
+      $ws= new WebSocket($this->server->connection, '/ws');
+      $ws->connect();
+      $ws->send($input);
+      $result= $ws->receive();
+    } finally {
+      $ws->close();
+    }
+    Assert::equals($output, $result);
+  }
+
+  #[Test, Expect(class: ProtocolException::class, message: 'Connection closed (#1007): Not valid utf-8')]
+  public function invalid_utf8_passed_to_websocket_text_message() {
+    try {
+      $ws= new WebSocket($this->server->connection, '/ws');
+      $ws->connect();
+      $ws->send("\xfc");
+      $ws->receive();
+    } finally {
+      $ws->close();
+    }
+  }
+
+  #[After]
+  public function shutdown() {
+    $this->server->shutdown();
+  }
 }

src/it/php/web/unittest/TestingApplication.class.php

@@ -1,14 +1,23 @@
 <?php namespace web\unittest;
 
+use Throwable;
 use lang\XPClass;
-use test\Assert;
+use util\Bytes;
+use web\handler\WebSocket;
 use web\{Application, Error};
 
 class TestingApplication extends Application {
 
   /** @return var */
   public function routes() {
     return [
+      '/ws' => new WebSocket(function($conn, $payload) {
+        if ($payload instanceof Bytes) {
+          $conn->send(new Bytes("\057\013{$payload}"));
+        } else {
+          $conn->send('Echo: '.$payload);
+        }
+      }),
       '/status/420' => function($req, $res) {
         $res->answer(420, $req->param('message') ?? 'Enhance your calm');
         $res->send('Answered with status 420', 'text/plain');
@@ -20,7 +29,7 @@ public function routes() {
       },
       '/raise/exception' => function($req, $res) {
         $class= XPClass::forName(basename($req->uri()->path()));
-        if ($class->isSubclassOf(\Throwable::class)) throw $class->newInstance('Raised');
+        if ($class->isSubclassOf(Throwable::class)) throw $class->newInstance('Raised');
 
         // A non-exception class was passed!
         $res->answer(200, 'No error');

src/it/php/web/unittest/TestingServer.class.php

@@ -5,7 +5,7 @@
 use peer\server\AsyncServer;
 use util\cmd\Console;
 use web\{Environment, Logging};
-use xp\web\srv\HttpProtocol;
+use xp\web\srv\{Protocol, HttpProtocol, WebSocketProtocol};
 
 /**
  * Socket server used by integration tests. 
@@ -23,10 +23,14 @@ class TestingServer {
   public static function main(array $args) {
     $application= new TestingApplication(new Environment('test', '.', '.', '.', [], null));
     $socket= new ServerSocket('127.0.0.1', $args[0] ?? 0);
+    $log= new Logging(null);
 
     $s= new AsyncServer();
     try {
-      $s->listen($socket, HttpProtocol::executing($application, new Logging(null)));
+      $s->listen($socket, Protocol::multiplex()
+        ->serving('http', new HttpProtocol($application, $log))
+        ->serving('websocket', new WebSocketProtocol(null, $log))
+      );
       $s->init();
       Console::writeLinef('+ Service %s:%d', $socket->host, $socket->port);
       $s->service();

src/main/php/web/Application.class.php

@@ -80,13 +80,14 @@ public function install($arg) {
    *
    * @param  web.Request $request
    * @param  web.Response $response
-   * @return var
+   * @return iterable
    */
   public function service($request, $response) {
     $seen= [];
 
     // Handle dispatching
     dispatch: $result= $this->routing()->handle($request, $response);
+    $return= null;
     if ($result instanceof Traversable) {
       foreach ($result as $kind => $argument) {
         if ('dispatch' === $kind) {
@@ -96,10 +97,16 @@ public function service($request, $response) {
             throw new Error(508, 'Internal redirect loop caused by dispatch to '.$argument);
           }
           goto dispatch;
+        } else if ('connection' === $kind) {
+          $response->header('Connection', 'upgrade');
+          $response->header('Upgrade', $argument[0]);
+          $return= $argument;
+        } else {
+          yield $kind => $argument;
         }
-        yield $kind => $argument;
       }
     }
+    return $return;
   }
 
   /** @return string */

src/main/php/web/Logging.class.php

@@ -58,15 +58,33 @@ public function tee($sink) {
   }
 
   /**
-   * Writes a log entry
+   * Writes an HTTP exchange to the log
    *
    * @param  web.Request $response
    * @param  web.Response $response
    * @param  [:var] $hints Optional hints
    * @return void
    */
-  public function log($request, $response, $hints= []) {
-    $this->sink && $this->sink->log($request, $response, $hints);
+  public function exchange($request, $response, $hints= []) {
+    $this->sink && $this->sink->log(
+      $response->status(),
+      $request->method(),
+      $request->uri()->resource(),
+      $response->trace + $hints
+    );
+  }
+
+  /**
+   * Writes a log entry
+   *
+   * @param  string $status
+   * @param  string $method
+   * @param  string $resource
+   * @param  [:var] $hints Optional hints
+   * @return void
+   */
+  public function log($status, $method, $resource, $hints= []) {
+    $this->sink && $this->sink->log($status, $method, $resource, $hints);
   }
 
   /**

src/main/php/web/Request.class.php

@@ -30,7 +30,7 @@ public function __construct(Input $input) {
     }
 
     $this->method= $input->method();
-    $this->uri= (new URI($input->scheme().'://'.$this->header('Host', 'localhost').$input->uri()))->canonicalize();
+    $this->uri= (new URI($input->scheme().'://'.$this->header('Host', 'localhost').$input->resource()))->canonicalize();
     $this->input= $input;
   }
 

src/main/php/web/handler/WebSocket.class.php

@@ -0,0 +1,70 @@
+<?php namespace web\handler;
+
+use web\Handler;
+use web\io\EventSink;
+use websocket\Listeners;
+
+/**
+ * WebSocket handler used for routing websocket handshake requests
+ *
+ * @test  web.unittest.handler.WebSocketTest
+ * @see   https://www.rfc-editor.org/rfc/rfc6455
+ */
+class WebSocket implements Handler {
+  const GUID= '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
+
+  private $listener;
+
+  /** @param function(websocket.protocol.Connection, string|util.Bytes): var|websocket.Listener $listener */
+  public function __construct($listener) {
+    $this->listener= Listeners::cast($listener);
+  }
+
+  /**
+   * Handles a request
+   *
+   * @param   web.Request $request
+   * @param   web.Response $response
+   * @return  var
+   */
+  public function handle($request, $response) {
+    switch ($version= (int)$request->header('Sec-WebSocket-Version')) {
+      case 13: // RFC 6455
+        $key= $request->header('Sec-WebSocket-Key');
+        $response->answer(101);
+        $response->header('Sec-WebSocket-Accept', base64_encode(sha1($key.self::GUID, true)));
+        foreach ($this->listener->protocols ?? [] as $protocol) {
+          $response->header('Sec-WebSocket-Protocol', $protocol, true);
+        }
+        break;
+
+      case 9: // Reserved version, use for WS <-> SSE translation
+        $response->answer(200);
+        $response->header('Content-Type', 'text/event-stream');
+        $response->header('Transfer-Encoding', 'chunked');
+        $response->trace('websocket', $request->header('Sec-WebSocket-Id'));
+
+        $events= new EventSink($request, $response);
+        foreach ($events->receive() as $message) {
+          $this->listener->message($events, $message);
+        }
+        return;
+
+      case 0:
+        $response->answer(426);
+        $response->send('This service requires use of the WebSocket protocol', 'text/plain');
+        return;
+
+      default:
+        $response->answer(400);
+        $response->send('This service does not support WebSocket version '.$version, 'text/plain');
+        return;
+    }
+
+    yield 'connection' => ['websocket', [
+      'path'     => $request->uri()->resource(),
+      'headers'  => $request->headers(),
+      'listener' => $this->listener,
+    ]];
+  }
+}

src/main/php/web/io/EventSink.class.php

@@ -0,0 +1,61 @@
+<?php namespace web\io;
+
+use io\streams\Streams;
+use lang\IllegalStateException;
+use util\Bytes;
+use websocket\protocol\{Opcodes, Connection};
+
+/** @test web.unittest.io.EventSinkTest */
+class EventSink extends Connection {
+  private $request, $out;
+
+  /**
+   * Creates a new event sink
+   *
+   * @param  web.Request $request
+   * @param  web.Response $response
+   */
+  public function __construct($request, $response) {
+    $this->request= $request;
+    $this->out= $response->stream();
+    parent::__construct(null, null, null, $request->uri()->resource(), $request->headers());
+  }
+
+  /**
+   * Receives messages
+   *
+   * @return iterable
+   */
+  public function receive() {
+    switch ($mime= $this->request->header('Content-Type')) {
+      case 'text/plain': yield Opcodes::TEXT => Streams::readAll($this->request->stream()); break;
+      case 'application/octet-stream': yield Opcodes::BINARY => new Bytes(Streams::readAll($this->request->stream())); break;
+      default: throw new IllegalStateException('Unexpected content type '.$mime);
+    }
+  }
+
+  /**
+   * Sends a websocket message
+   *
+   * @param  string|util.Bytes $message
+   * @return void
+   */
+  public function send($message) {
+    if ($message instanceof Bytes) {
+      $this->out->write("event: bytes\ndata: ".addcslashes($message, "\r\n")."\n\n");
+    } else {
+      $this->out->write("data: ".addcslashes($message, "\r\n")."\n\n");
+    }
+  }
+
+  /**
+   * Closes the websocket connection
+   *
+   * @param  int $code
+   * @param  string $reason
+   * @return void
+   */
+  public function close($code= 1000, $reason= '') {
+    $this->out->write("event: close\ndata: ".$code.':'.addcslashes($reason, "\r\n")."\n\n");
+  }
+}