Fixed some phpstan errors

README.md

@@ -1,18 +1,31 @@
 # ReactPHP WebSockets
 
+This is a library that implements WebSocket support on top of ReactPHPs HttpServer. In the process, the underlaying request and endpoint info is made available to your code. What happens next is up to you.
+
+## Missing Features
+
+This library is under development, and should not be used in any projects yet. The API and implementation may change in the future, and some features are still not implemented, or work in progress:
+
+* Timeouts -- Disconnect the client on ping/pong timeout, and optionally on inactivity/idle.
+* Error Handling -- Protocol errors should close the connection etc.
+* Exceptions -- Exceptions should be thrown when relevant errors are encountered, such as bad encoding or data.
+* Fragmented Messages -- Sending and receiving messages fragmented over multiple frames need work.
+
 ## Installing
 
 Install using composer:
 
 ```shell
-$ composer require noccylabs/react-websocket
+$ composer require noccylabs/react-websocket:^0.1.0
 ```
 
-## Why not Ratchet?
+Note that you may need additional permissions to be able to access the repository. If so, clone it locally:
 
-Ratchet is great! I've used Ratchet in the past, and it is a fantastic piece of code. It is however more application-centered, which means it doesn't do events and all that beautiful magic we've come to love ReactPHP for.
-
-TL;DR - If you need to build an application with neatly wrapped classes without caring to much about the internals, go with Ratchet. If you want to work with websockets in the same way you work with sockets in ReactPHP, go with this library.
+```shell
+$ git clone https://dev.noccylabs.info/noccy/react-websocket ~/react-websocket
+$ composer config repositories.react-websocket vcs ~/react-websocket
+$ composer require noccylabs/react-websocket:^0.1.0 # ^0.1.0 or @dev
+```
 
 ## Server
 
@@ -55,30 +68,86 @@ $http->listen($socket);
 
 ```
 
-### Server Events
+### WebSocketMiddleware Events
 
 #### connection
 
+```php
+function (WebSocketInterface $member)
+```
+
 This event is emitted when a new WebSocket request has been accepted. The `WebSocketConnection` is passed as the first argument.
 
 ### WebSocketConnection events
 
 #### ping
 
+```php
+function (string $payload)
+```
+
 This event will be emitted upon receiving a frame with a ping opcode. The pong response has already been sent automatically, unless 'no_auto_pong' is set in the context.
 
 #### pong
 
+```php
+function (string $payload)
+```
+
 This event will be emitted upon receiving a frame with a pong opcode.
 
 #### text
 
+```php
+function (string $payload)
+```
+
 This event will be emitted when a text data frame have been received and decoded.
 
 #### binary
 
+```php
+function (string $payload)
+```
+
 This event will be emitted when a binary data frame have been received and decoded.
 
 #### close
 
+```php
+function ()
+```
+
 #### error
+
+```php
+function (?string $reason, ?int $code)
+```
+
+### GroupManager events
+
+#### create
+
+```php
+function (ConnectionGroup $group)
+```
+
+#### destroy
+
+```php
+function (ConnectionGroup $group)
+```
+
+### ConnectionGroup events
+
+#### join
+
+```php
+function (WebSocketInterface $member)
+```
+
+#### leave
+
+```php
+function (WebSocketInterface $member)
+```

doc/ConnectionGroups.md

@@ -36,6 +36,14 @@ $groupManager->on('created', function (ConnectionGroup $group) {
 $middleware = new WebSocketMiddleware($groupManager);
 ```
 
+## Sending messages
+
+You can use the `ConnectionGoup::writeAll(string $payload)` method to send the payload to all members of the group.
+
+## Disconnecting clients
+
+You can disconnect clients cleanly on shutdown by using the `GroupManager::closeAll(string $reason, int $code)` method. You can also call on `ConnectionGrroup::closeAll` manually do disconnect a whole group.
+
 ## Future
 
 * Add a GroupManagerImplementation so custom logic can be provided.

doc/ObjectUserdata.md

@@ -0,0 +1,5 @@
+# Object UserData
+
+There is no support for userdata on `WebSocketInterface`, `WebSocketConnection` or `ConnectionGroup`.
+
+The rationale for this is that the connections can be easily managed using `SplObjectStorage` to link to a data object. Similarly, for groups, the names should be unique and can be used for lookups.

examples/server.php

@@ -10,25 +10,34 @@ use React\Http\Message\Response;
 use React\Promise\Promise;
 use React\Socket\SocketServer;
 
+// The middleware is at the core, as it intercepts and breaks out the websocket
+// connections received by the HttpServer.
 $websockets = new WebSocketMiddleware();
+// The 'connection' handler works like expected. You can't however write any data
+// to the client yet, as the websocket streams have not yet been returned.
 $websockets->on('connection', function (WebSocketInterface $websocket) {
+    // This just echoes text received, unless the websocket is part of a group.
+    // In this case the message is sent to all websockets in the group.
     $websocket->on('text', function ($text) use ($websocket) {
+        // Send a message with the group name starting with '#' to join a group.
         if (str_starts_with($text, '#')) {
             $websocket->setGroup(substr($text,1));
             $websocket->write("joined group {$text}");
         } else {
+            // Echo back if not in group, send to group otherwise
             if (!$websocket->getGroup())
                 $websocket->write($text);
             else
-            foreach ($websocket->getGroup() as $member) {
+            foreach ($websocket->getGroup() as $member)
                 $member->write($text);
-            }
         }
     });
 });
 
+// Create the server with our middleware chain:
+//   error handler -> websocket handler -> http handler
 $server = new HttpServer(
-    $websockets,
+    // Error handler
     function (ServerRequestInterface $request, callable $next) {
         $promise = new Promise(function ($resolve) use ($next, $request) {
             $resolve($next($request));
@@ -39,11 +48,14 @@ $server = new HttpServer(
             )->withStatus(Response::STATUS_INTERNAL_SERVER_ERROR);
         });
     },
+    // WebSocket handler
+    $websockets,
+    // HTTP handler
     function (ServerRequestInterface $request) {
         return Response::plaintext("Hello world!");
     }
 );
 
+// Everything else is as expected!
 $socket = new SocketServer("tcp://0.0.0.0:8000");
-
-$server->listen($socket);
+$server->listen($socket);

src/Debug/HexDumpTrait.php

@@ -1,28 +0,0 @@
-<?php
-
-namespace NoccyLabs\React\WebSocket\Debug;
-
-trait HexDumpTrait
-{
-
-    private function hexdump($data): void
-    {
-        printf("_____  *%4d\n", strlen($data));
-        $rows = str_split($data, 16);
-        $offs = 0;
-        foreach ($rows as $row) {
-            $h = []; $a = [];
-            for ($n = 0; $n < 16; $n++) {
-                if ($n < strlen($row)) {
-                    $h[] = sprintf("%02x%s", ord($row[$n]), ($n==7)?"  ":" ");
-                    $a[] = sprintf("%s%s", (ctype_print($row[$n])?$row[$n]:"."), ($n==7)?" ":"");
-                } else {
-                    $h[] = (($n==7)?"    ":"   ");
-                    $a[] = (($n==7)?"  ":" ");
-                }
-            }
-            printf("%04x | %s | %s\n", 16 * $offs++, join("", $h), join("", $a));
-        }
-    }
-
-}

src/Group/ConnectionGroup.php

@@ -46,11 +46,19 @@ class ConnectionGroup implements EventEmitterInterface, IteratorAggregate, Count
         return count($this->connections);
     }
 
+    /** 
+     * 
+     * @return \Traversable<int,WebSocketInterface> 
+     */
     public function getIterator(): Traversable
     {
         return new ArrayIterator($this->connections);
     }
 
+    /**
+     * 
+     * @return WebSocketInterface[]
+     */
     public function getAll(): array
     {
         return $this->connections;
@@ -61,8 +69,24 @@ class ConnectionGroup implements EventEmitterInterface, IteratorAggregate, Count
         return $this->name;
     }
 
-    public function write(string $payload)
+    public function writeAll(string $payload)
     {
-        
+        foreach ($this->connections as $connection) {
+            if ($connection->isWritable()) {
+                $connection->write($payload);
+            }
+        }
     }
+
+    /**
+     * 
+     * @return void
+     */
+    public function closeAll(string $reason, int $code=1001)
+    {
+        foreach ($this->connections as $connection) {
+            $connection->closeWithReason($reason, $code);
+        }
+    }
+
 }

src/Group/GroupManager.php

@@ -13,11 +13,11 @@ class GroupManager implements EventEmitterInterface
     /**
      * @var string emitted when a new group is created
      */
-    const EVENT_CREATED = 'created';
+    const EVENT_CREATE = 'create';
     /**
      * @var string emitted after the last member leaves, when the group is destroyed
      */
-    const EVENT_DESTROYED = 'destroyed';
+    const EVENT_DESTROY = 'destroy';
 
     /** @var array<string,ConnectionGroup> */
     private array $groups = [];
@@ -31,18 +31,24 @@ class GroupManager implements EventEmitterInterface
             $group->on(ConnectionGroup::EVENT_LEAVE, function () use ($group) {
                 Loop::futureTick(function () use ($group) {
                     if (count($group) === 0) {
-                        $this->emit(self::EVENT_DESTROYED, [ $group ]);
+                        $this->emit(self::EVENT_DESTROY, [ $group ]);
                         $group->removeAllListeners();
                         unset($this->groups[$group->getName()]);
                     }
                 });
             });
-            $this->emit(self::EVENT_CREATED, [ $group ]);
+            $this->emit(self::EVENT_CREATE, [ $group ]);
         } else {
             $group = $this->groups[$name];
         }
         return $group;
     }
 
+    public function closeAll(string $reason, int $code=1001)
+    {
+        foreach ($this->groups as $group) {
+            $group->closeAll($reason, $code);
+        }
+    }
     
 }

src/WebSocketConnection.php

@@ -18,6 +18,7 @@ class WebSocketConnection implements WebSocketInterface
 {
     use EventEmitterTrait;
 
+    // TODO maybe move constants to WebSocketProtocol?
     const OP_CONTINUATION = 0x0;
     const OP_FRAME_TEXT = 0x1;
     const OP_FRAME_BINARY = 0x2;
@@ -28,8 +29,8 @@ class WebSocketConnection implements WebSocketInterface
     /** @var string|null The name of the group that this connection has joined, or null */
     private ?string $groupName = null;
 
-    /** @var WebSocketCodec The frame encoder/decoder */
-    private WebSocketCodec $codec;
+    /** @var WebSocketProtocol The frame encoder/decoder */
+    private WebSocketProtocol $codec;
 
     private ?ConnectionGroup $group = null;
 
@@ -41,14 +42,16 @@ class WebSocketConnection implements WebSocketInterface
 
     private ServerRequestInterface $request;
 
+    /** @var string|null Buffer for fragmented messages */
     private ?string $buffer = null;
 
+    /** @var int|null The opcode of a fragmented message */
     private ?int $bufferedOp = null;
 
     public function __construct(ServerRequestInterface $request, ReadableStreamInterface $inStream, WritableStreamInterface $outStream, GroupManager $groupManager)
     {
         // The codec is used to encode and decode frames
-        $this->codec = new WebSocketCodec();
+        $this->codec = new WebSocketProtocol();
         
         $this->request = $request;
         $this->inStream = $inStream;
@@ -56,10 +59,7 @@ class WebSocketConnection implements WebSocketInterface
         $this->groupManager = $groupManager;
 
         $this->inStream->on('data', $this->onWebSocketData(...));
-        $this->inStream->on('close', function () {
-            $this->close();
-            $this->emit('close', []);
-        });
+        $this->inStream->on('close', $this->close(...));
     }
 
     private function onWebSocketData($data)
@@ -77,15 +77,15 @@ class WebSocketConnection implements WebSocketInterface
             } else {
                 $this->buffer .= $payload;
             }
+            // Break out to avoid processing partial messages
+            return;
         }
 
-        if ($final) {
+        if ($this->bufferedOp !== null) {
             $payload = $this->buffer . $payload;
             $this->buffer = null;
-            if ($this->bufferedOp !== null) {
-                $opcode = $this->bufferedOp;
-                $this->bufferedOp = null;
-            }
+            $opcode = $this->bufferedOp;
+            $this->bufferedOp = null;
         }
 
         switch ($opcode) {
@@ -163,6 +163,11 @@ class WebSocketConnection implements WebSocketInterface
         return $this->group;
     }
 
+    public function getServerRequest(): ServerRequestInterface
+    {
+        return $this->request;
+    }
+
     public function getRemoteAddress()
     {
         return $this->request->getServerParams()['REMOTE_ADDR'];
@@ -216,19 +221,33 @@ class WebSocketConnection implements WebSocketInterface
 
     /**
      * {@inheritDoc}
+     * 
+     * @see writeBinary() to write binary frames as opposed to text frames.
      */
     public function write($data)
     {           
         return $this->send(self::OP_FRAME_TEXT, $data);
     }
 
+    /**
+     * Write binary frames.
+     * 
+     * @param string $data
+     * @return bool
+     */
     public function writeBinary($data)
     {           
         return $this->send(self::OP_FRAME_BINARY, $data);
     }
 
     /**
+     * Encode and send a frame.
      * 
+     * @param int $opcode
+     * @param string $data
+     * @param bool $final
+     * @param bool $masked
+     * @return bool
      */
     public function send(int $opcode, string $data, bool $final = true, bool $masked = false)
     {
@@ -240,9 +259,7 @@ class WebSocketConnection implements WebSocketInterface
             'masked' => $masked
         ]);
 
-        $this->outStream->write($frame);
-
-        return true;
+        return $this->outStream->write($frame);
     }
 
     /**
@@ -252,6 +269,17 @@ class WebSocketConnection implements WebSocketInterface
     {
         $this->outStream->close();
         $this->inStream->close();
+        
+        $this->emit('close', []);        
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function closeWithReason(string $reason, int $code=1000): void
+    {
+        $payload = chr(($code >> 8) & 0xFF) . chr($code & 0xFF) . $reason;
+        $this->send(self::OP_CLOSE, $payload);
     }
 
     /**
@@ -259,7 +287,7 @@ class WebSocketConnection implements WebSocketInterface
      */
     public function end($data = null)
     {
-        
+        // TODO implement me
     }
 
 }

src/WebSocketInterface.php

@@ -16,11 +16,46 @@ interface WebSocketInterface extends ConnectionInterface
     const EVENT_GROUP_JOIN = 'join';
     const EVENT_GROUP_LEAVE = 'leave';
 
+    /**
+     * Close the connection with a reason and code.
+     * 
+     * @param string $reason
+     * @param int $code
+     * @return void
+     */
+    public function closeWithReason(string $reason, int $code=1000): void;
 
+    /**
+     * Get the initial HTTP request sent to the server.
+     * 
+     * @return ServerRequestInterface
+     */
+    public function getServerRequest(): ServerRequestInterface;
+
+    /**
+     * Assign this connection to a connection group. If the connection is already
+     * part of a group, it will leave the current group before joining the new
+     * group.
+     * 
+     * @param null|string $name The group name to join
+     * @return void
+     */
     public function setGroup(?string $name): void;
 
-    public function getGroupName(): ?string;
-
+    /**
+     * Get the current connection group.
+     * 
+     * @see getGroupName() if you want the name of the group.
+     * 
+     * @return null|ConnectionGroup
+     */
     public function getGroup(): ?ConnectionGroup;
 
-}
+    /**
+     * Get the name of the current connection group
+     * 
+     * @return null|string
+     */
+    public function getGroupName(): ?string;
+
+}

src/WebSocketMiddleware.php

@@ -18,6 +18,8 @@ class WebSocketMiddleware implements EventEmitterInterface
 
     const MAGIC = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
 
+    const VERSION = 13;
+
     use EventEmitterTrait;
 
     private GroupManager $groupManager;
@@ -32,7 +34,7 @@ class WebSocketMiddleware implements EventEmitterInterface
 
     public function addRoute(string $path, callable $handler, array $allowedOrigins=[]): void
     {
-
+        // TODO implement or remove
     }
 
     public function __invoke(ServerRequestInterface $request, callable $next)
@@ -62,12 +64,15 @@ class WebSocketMiddleware implements EventEmitterInterface
             $this->emit(self::EVENT_CONNECTION, [ $websocket ]);
         //});
 
+        // TODO would it be possible for the 'connection' event to set additional response headers to be sent here?
+        // For example, to send back Sec-WebSocket-Protocol header.
         return new Response(
             Response::STATUS_SWITCHING_PROTOCOLS,
             array(
                 'Upgrade' => 'websocket',
                 'Connection' => 'upgrade',
-                'Sec-WebSocket-Accept' => $handshakeResponse
+                'Sec-WebSocket-Accept' => $handshakeResponse,
+                'Sec-WebSocket-Version' => self::VERSION
             ),
             $stream
         );        

src/WebSocketProtocol.php

@@ -14,7 +14,7 @@ use React\Stream\ReadableStreamInterface;
 use React\Stream\ThroughStream;
 use React\Stream\WritableStreamInterface;
 
-class WebSocketCodec
+class WebSocketProtocol
 {
 
     /**
@@ -111,7 +111,7 @@ class WebSocketCodec
      *   payload (string) - the payload
      * 
      * @param string $frame The frame data to decode
-     * @return array The decoded frame
+     * @return array<string,mixed> The decoded frame
      */
     public function decode(string $frame): array
     {
@@ -125,6 +125,7 @@ class WebSocketCodec
         $byte0 = ord($frame[0]);
         $decoded['final'] = !!($byte0 & 0x80);
         $decoded['opcode'] = $byte0 & 0x0F;
+
         // Peek at the second byte, holding mask bit and len
         $byte1 = ord($frame[1]);
         $decoded['masked'] = $masked = !!($byte1 & 0x80);
@@ -146,14 +147,16 @@ class WebSocketCodec
             $header += 4;
         }
 
-        // Now for the masking
+        // Now for the masking, if present.
         if ($masked) {
             $mask = substr($frame, $header, 4);
             $header += 4;
         }
 
-        // Extract and unmask payload
+        // Extract the payload, and unmask it if needed. The mask() function handles
+        // both masking and unmasing as the algorithm uses xor.
         $payload = substr($frame, $header, $len);
+        // TODO check that extracted payload len equals expected len
         if ($masked) {
             $payload = $this->mask($payload, $mask);
             $decoded['mask'] = $mask;
@@ -174,14 +177,17 @@ class WebSocketCodec
      */
     private function mask(string $payload, string $mask): string
     {
+        // Unpack the payload and mask into byte values
         $payloadData = array_map("ord", str_split($payload,1));
         $maskData = array_map("ord", str_split($mask,1));
+        // TODO check that mask len==4 
 
         $unmasked = [];
         for ($n = 0; $n < count($payloadData); $n++) {
             $unmasked[] = $payloadData[$n] ^ $maskData[$n % 4];
         }
 
+        // Return the masked byte values packed into a string
         return join("", array_map("chr", $unmasked));
     }
 

tests/Group/ConnectionGroupTest.php

@@ -0,0 +1,55 @@
+<?php
+
+namespace NoccyLabs\React\WebSocket\Group;
+
+use NoccyLabs\React\WebSocket\WebSocketConnection;
+use NoccyLabs\React\WebSocket\WebSocketProtocol;
+use PHPUnit\Framework\Attributes\CoversClass;
+use React\Http\Message\ServerRequest;
+use React\Stream\ThroughStream;
+
+#[CoversClass(ConnectionGroup::class)]
+#[CoversClass(WebSocketConnection::class)]
+#[CoversClass(WebSocketProtocol::class)]
+class ConnectionGroupTest extends \PHPUnit\Framework\TestCase
+{
+
+    public function testThatGroupsAreCountable()
+    {
+        $groupManager = new GroupManager();
+        
+        $group = new ConnectionGroup();
+        $this->assertEquals(0, count($group));
+        $group->add(new WebSocketConnection(new ServerRequest('GET','/'),new ThroughStream(), new ThroughStream(), $groupManager));
+        $this->assertEquals(1, count($group));
+        $group->add(new WebSocketConnection(new ServerRequest('GET','/'),new ThroughStream(), new ThroughStream(), $groupManager));
+        $this->assertEquals(2, count($group));
+        $group->add(new WebSocketConnection(new ServerRequest('GET','/'),new ThroughStream(), new ThroughStream(), $groupManager));
+        $this->assertEquals(3, count($group));
+    }
+
+    public function testThatGroupNamesAreAssignedUniquely()
+    {
+        $group = new ConnectionGroup();
+        $this->assertNotEmpty($group->getName());
+
+        $group2 = new ConnectionGroup();
+        $this->assertNotEmpty($group2->getName());
+
+        $this->assertNotEquals($group->getName(), $group2->getName());
+    }
+
+    public function testWritingToAll()
+    {
+        $groupManager = new GroupManager();
+        
+        $r = [];
+        $group = new ConnectionGroup();
+        $group->add(new WebSocketConnection(new ServerRequest('GET','/'),new ThroughStream(), new ThroughStream(function ($data) use (&$r) { $r[]=$data; }), $groupManager));
+        $group->add(new WebSocketConnection(new ServerRequest('GET','/'),new ThroughStream(), new ThroughStream(function ($data) use (&$r) { $r[]=$data; }), $groupManager));
+        $group->add(new WebSocketConnection(new ServerRequest('GET','/'),new ThroughStream(), new ThroughStream(function ($data) use (&$r) { $r[]=$data; }), $groupManager));
+        $group->writeAll("test");
+        $this->assertEquals(3, count($r));
+    }
+
+}

tests/WebSocketMiddlewareTest.php

@@ -6,6 +6,7 @@ use PHPUnit\Framework\Attributes\CoversClass;
 use React\Http\Message\ServerRequest;
 
 #[CoversClass(WebSocketMiddleware::class)]
+#[CoversClass(WebSocketConnection::class)]
 class WebSocketMiddlewareTest extends \PHPUnit\Framework\TestCase
 {
 
@@ -54,4 +55,4 @@ class InvokableDummy
     public function test()
     {
     }
-}
+}

tests/WebSocketProtocolTest.php

@@ -4,13 +4,13 @@ namespace NoccyLabs\React\WebSocket;
 
 use PHPUnit\Framework\Attributes\CoversClass;
 
-#[CoversClass(WebSocketCodec::class)]
-class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
+#[CoversClass(WebSocketProtocol::class)]
+class WebSocketProtocolTest extends \PHPUnit\Framework\TestCase
 {
 
     public function testEncodingFrames()
     {
-        $codec = new WebSocketCodec();
+        $codec = new WebSocketProtocol();
 
         $msg = $codec->encode([
             'opcode'=>WebSocketConnection::OP_PING,
@@ -43,7 +43,7 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
 
     public function testDecodingFrames()
     {
-        $codec = new WebSocketCodec();
+        $codec = new WebSocketProtocol();
 
         $msg = $codec->decode("\x89\x04ping");
         $this->assertEquals([