Updated readme

README.md

@@ -14,6 +14,14 @@ Ratchet is great! I've used Ratchet in the past, and it is a fantastic piece of
 
 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.
 
+## Missing Features
+
+The following features are missing, or work in progress:
+
+* Idle timeout and ping timeout
+* Protocol errors should close with error codes
+* Exceptions
+
 ## Server
 
 The WebSocket handler is built as a HttpServer middleware. This makes sense as WebSocket as a protocol is running over HTTP. Connections are set up by the middleware and exposed via the `connect` event.
@@ -55,30 +63,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)
+```

composer.json

@@ -3,6 +3,7 @@
     "description": "Native ReactPHP WebSocket implementation",
     "type": "library",
     "license": "GPL-3.0-or-later",
+    "keywords": [ "reactphp", "websockets" ],
     "autoload": {
         "psr-4": {
             "NoccyLabs\\React\\WebSocket\\": "src/"
@@ -18,6 +19,7 @@
         "react/http": "^1.9.0"
     },
     "require-dev": {
-        "phpunit/phpunit": "^11.0"
+        "phpunit/phpunit": "^11.0",
+        "phpstan/phpstan": "^1.10"
     }
 }

doc/ConnectionGroups.md

@@ -15,4 +15,36 @@ foreach ($websocket->getGroup() as $other) {
 
 To remove a group from a connection, pass `null` to `WebSocketConnection::setGroup()`.
 
-The group will emit a `join` event (`WebSocketConnection::EVENT_JOIN`) when another member joins the group, and a `leave` event (`WebSocketConnection::EVENT_LEAVE`) when a member leaves. The events will be sent to the leaving member as well, so consider this in your logic.
+The group will emit a `join` event (`ConnectionGroup::EVENT_JOIN`) when another member joins the group, and a `leave` event (`ConnectionGroup::EVENT_LEAVE`) when a member leaves. The events will be sent to the leaving member as well, so consider this in your logic.
+
+## Events
+
+The GroupManager emits events when a group is `created` (`GroupManager::EVENT_CREATED`) or `destroyed` (`GroupManager::EVENT_DESTROYED`). You can use these events to hook the join and leave events.
+
+```php
+// Create a GroupManager
+$groupManager = new GroupManager();
+$groupManager->on('created', function (ConnectionGroup $group) {
+  // Listen for joins
+  $group->on('join', function (WebSocketConnection $connection) use ($group) {
+    // Someone joined the group!
+    $group->write("Someone joined!");
+  })
+});
+
+// The GroupManager is injected into the WebSocketMiddleware
+$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.
+* Make it possible to reject setting a group by GroupManager not returning a group.

phpstan.neon

@@ -0,0 +1,12 @@
+parameters:
+  level: 5
+    
+  excludePaths:
+    - doc
+    - vendor
+    - tests
+
+  # Paths to include in the analysis
+  paths:
+    - src
+

src/Debug/HexDumpTrait.php

@@ -0,0 +1,28 @@
+<?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

@@ -60,4 +60,21 @@ class ConnectionGroup implements EventEmitterInterface, IteratorAggregate, Count
     {
         return $this->name;
     }
+
+    public function writeAll(string $payload)
+    {
+        foreach ($this->connections as $connection) {
+            if ($connection->isWritable()) {
+                $connection->write($payload);
+            }
+        }
+    }
+
+    public function closeAll(string $reason, int $code=1001)
+    {
+        foreach ($this->connections as $connection) {
+            $connection->closeWithReason($reason, $code);
+        }
+    }
+
 }

src/Group/GroupManager.php

@@ -2,12 +2,23 @@
 
 namespace NoccyLabs\React\WebSocket\Group;
 
-use NoccyLabs\React\WebSocket\WebSocketInterface;
+use Evenement\EventEmitterInterface;
+use Evenement\EventEmitterTrait;
 use React\EventLoop\Loop;
-use WeakReference;
 
-class GroupManager
+class GroupManager implements EventEmitterInterface
 {
+    use EventEmitterTrait;
+
+    /**
+     * @var string emitted when a new group is created
+     */
+    const EVENT_CREATE = 'create';
+    /**
+     * @var string emitted after the last member leaves, when the group is destroyed
+     */
+    const EVENT_DESTROY = 'destroy';
+
     /** @var array<string,ConnectionGroup> */
     private array $groups = [];
 
@@ -20,16 +31,24 @@ class GroupManager
             $group->on(ConnectionGroup::EVENT_LEAVE, function () use ($group) {
                 Loop::futureTick(function () use ($group) {
                     if (count($group) === 0) {
+                        $this->emit(self::EVENT_DESTROY, [ $group ]);
                         $group->removeAllListeners();
                         unset($this->groups[$group->getName()]);
                     }
                 });
             });
+            $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

@@ -25,9 +25,11 @@ class WebSocketConnection implements WebSocketInterface
     const OP_PING = 0x9;
     const OP_PONG = 0xA;
 
+    /** @var string|null The name of the group that this connection has joined, or null */
     private ?string $groupName = null;
 
-    private WebSocketCodec $codec;
+    /** @var WebSocketProtocol The frame encoder/decoder */
+    private WebSocketProtocol $codec;
 
     private ?ConnectionGroup $group = null;
 
@@ -39,14 +41,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;
@@ -54,6 +58,10 @@ class WebSocketConnection implements WebSocketInterface
         $this->groupManager = $groupManager;
 
         $this->inStream->on('data', $this->onWebSocketData(...));
+        $this->inStream->on('close', function () {
+            $this->close();
+            $this->emit('close', []);
+        });
     }
 
     private function onWebSocketData($data)
@@ -87,6 +95,7 @@ class WebSocketConnection implements WebSocketInterface
                 $this->sendPong($payload);
                 return;
             case self::OP_PONG:
+                $this->checkPong($payload);
                 return;
             case self::OP_CLOSE:
                 // TODO implement
@@ -103,11 +112,27 @@ class WebSocketConnection implements WebSocketInterface
         }
     }
 
+    /**
+     * Sends a ping, and closes the connection on timeout.
+     * 
+     */
+    public function ping(): void
+    {
+        // TODO save the state somehow
+        $payload = "ping";
+        $this->send(self::OP_PING, $payload, true);
+    }
+
     private function sendPong(string $data): void
     {
         $this->send(self::OP_PONG, $data, true);
     }
 
+    private function checkPong(string $data): void
+    {
+        // TODO reset the state and any ping timers
+    }
+
     public function setGroup(?string $name): void
     {
         if ($this->group) {
@@ -150,32 +175,50 @@ class WebSocketConnection implements WebSocketInterface
         return $this->request->getServerParams()['SERVER_ADDR'];
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function isReadable()
     {
         return $this->inStream->isReadable();
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function pause()
     {
-        return $this->inStream->pause();
+        $this->inStream->pause();
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function resume()
     {
-        return $this->inStream->resume();        
+        $this->inStream->resume();        
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function isWritable()
     {
         return $this->outStream->isWritable();        
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function pipe(WritableStreamInterface $dest, array $options = array())
     {
         // TODO implement
         return $dest;
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function write($data)
     {           
         return $this->send(self::OP_FRAME_TEXT, $data);
@@ -204,35 +247,30 @@ class WebSocketConnection implements WebSocketInterface
         return true;
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function close()
     {
+        // TODO send close
         $this->outStream->close();
         $this->inStream->close();
+        // TODO emit close event
     }
 
+    public function closeWithReason(string $reason, int $code=1000)
+    {
+        // TODO send close
+        $payload = chr(($code >> 8) & 0xFF) . chr($code & 0xFF) . $reason;
+        $this->send(self::OP_CLOSE, $payload);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
     public function end($data = null)
     {
         
     }
 
-    // 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/WebSocketInterface.php

@@ -23,4 +23,6 @@ interface WebSocketInterface extends ConnectionInterface
 
     public function getGroup(): ?ConnectionGroup;
 
+    public function closeWithReason(string $reason, int $code=1000);
+
 }

src/WebSocketProtocol.php

@@ -14,22 +14,29 @@ use React\Stream\ReadableStreamInterface;
 use React\Stream\ThroughStream;
 use React\Stream\WritableStreamInterface;
 
-class WebSocketCodec
+class WebSocketProtocol
 {
 
-    const OP_CONTINUATION = 0x0;
-    const OP_FRAME_TEXT = 0x1;
-    const OP_FRAME_BINARY = 0x2;
-    const OP_CLOSE = 0x8;
-    const OP_PING = 0x9;
-    const OP_PONG = 0xA;
-
+    /**
+     * Encode a frame. Required keys are opcode and payload. Keys that can be passed are:
+     * 
+     *   opcode (int) - the opcode
+     *   final (bool) - final frame
+     *   rsv1-3 (bool) - reserved bits
+     *   masked (bool) - if the frame was masked
+     *   mask (string) - the mask bytes, if masked
+     *   length (int) - length of payload
+     *   payload (string) - the payload
+     * 
+     * @param array $frame The frame
+     * @return string The encoded frame
+     */
     public function encode(array $frame): string
     {
         // Encoded frame
         $encoded = null;
 
-        // Unpack frame with defaults
+        // Re-unpack frame with defaults
         $frame = [
             ...[
                 'final' => true,
@@ -56,10 +63,13 @@ class WebSocketCodec
             $size1 = ($len >> 8) & 0xFF;
             $size2 = $len & 0xFF;
             $size3 = null;
+            $size4 = null;
         } else {
             $size0 = $len;
             $size1 = null;
+            $size2 = null;
             $size3 = null;
+            $size4 = null;
         }
 
         $encoded .= chr(($frame['final']?0x80:0x00) 
@@ -86,21 +96,24 @@ class WebSocketCodec
             $encoded .= $frame['payload'];
         }
 
-        //$this->hexdump($encoded);
-
         return $encoded;
     }
 
     /**
      * Decode a websocket frame and return an array with the keys:
+     * 
      *   opcode (int) - the opcode
-     *   fin (bool) - final frame
+     *   final (bool) - final frame
      *   rsv1-3 (bool) - reserved bits
      *   masked (bool) - if the frame was masked
+     *   mask (string) - the mask bytes, if masked
      *   length (int) - length of payload
      *   payload (string) - the payload
+     * 
+     * @param string $frame The frame data to decode
+     * @return array The decoded frame
      */
-    public function decode($frame): array
+    public function decode(string $frame): array
     {
         // Decoded frame
         $decoded = [];
@@ -152,38 +165,24 @@ class WebSocketCodec
         return $decoded;
     }
 
+    /**
+     * Masking is reversible, and simply xors a repeated 4-byte key with the data.
+     * 
+     * @param string $payload The unmasked (or masked) input
+     * @param string $mask The mask to use (4 bytes)
+     * @return string The masked (or unmasked) output
+     */
     private function mask(string $payload, string $mask): string
     {
         $payloadData = array_map("ord", str_split($payload,1));
         $maskData = array_map("ord", str_split($mask,1));
 
-        //printf("Mask: %02x %02x %02x %02x\n", ...$maskData);
-
         $unmasked = [];
         for ($n = 0; $n < count($payloadData); $n++) {
             $unmasked[] = $payloadData[$n] ^ $maskData[$n % 4];
         }
+
         return join("", array_map("chr", $unmasked));
     }
 
-    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));
-        }
-    }
-
 }

tests/Group/ConnectionGroupTest.php

@@ -0,0 +1,52 @@
+<?php
+
+namespace NoccyLabs\React\WebSocket\Group;
+
+use NoccyLabs\React\WebSocket\WebSocketConnection;
+use PHPUnit\Framework\Attributes\CoversClass;
+use React\Http\Message\ServerRequest;
+use React\Stream\ThroughStream;
+
+#[CoversClass(ConnectionGroup::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/WebSocketProtocolTest.php

@@ -4,27 +4,27 @@ 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'=>WebSocketCodec::OP_PING,
+            'opcode'=>WebSocketConnection::OP_PING,
             'payload'=>"ping"
         ]);
         $this->assertEquals("\x89\x04ping", $msg);
 
         $msg = $codec->encode([
-            'opcode'=>WebSocketCodec::OP_FRAME_TEXT,
+            'opcode'=>WebSocketConnection::OP_FRAME_TEXT,
             'payload'=>"abcdefgh"]);
         $this->assertEquals("\x81\x08abcdefgh", $msg);
 
         $msg = $codec->encode([
-            'opcode'=>WebSocketCodec::OP_FRAME_TEXT,
+            'opcode'=>WebSocketConnection::OP_FRAME_TEXT,
             'payload'=>"abcdefgh",
             'masked'=>true,
             'mask'=>"\x00\x00\x00\x00"
@@ -32,7 +32,7 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
         $this->assertEquals("\x81\x88\x00\x00\x00\x00abcdefgh", $msg);
 
         $msg = $codec->encode([
-            'opcode'=>WebSocketCodec::OP_FRAME_TEXT,
+            'opcode'=>WebSocketConnection::OP_FRAME_TEXT,
             'payload'=>"abcdefgh",
             'masked'=>true,
             'mask'=>"\x00\xFF\x00\xFF"
@@ -43,11 +43,11 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
 
     public function testDecodingFrames()
     {
-        $codec = new WebSocketCodec();
+        $codec = new WebSocketProtocol();
 
         $msg = $codec->decode("\x89\x04ping");
         $this->assertEquals([
-            'opcode'=>WebSocketCodec::OP_PING,
+            'opcode'=>WebSocketConnection::OP_PING,
             'payload'=>"ping",
             'final'=>true,
             'rsv1'=>false,
@@ -59,7 +59,7 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
 
         $msg = $codec->decode("\x81\x08abcdefgh");
         $this->assertEquals([
-            'opcode'=>WebSocketCodec::OP_FRAME_TEXT,
+            'opcode'=>WebSocketConnection::OP_FRAME_TEXT,
             'payload'=>"abcdefgh",
             'final'=>true,
             'rsv1'=>false,
@@ -71,7 +71,7 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
 
         $msg = $codec->decode("\x81\x88\x00\x00\x00\x00abcdefgh");
         $this->assertEquals([
-            'opcode'=>WebSocketCodec::OP_FRAME_TEXT,
+            'opcode'=>WebSocketConnection::OP_FRAME_TEXT,
             'payload'=>"abcdefgh",
             'final'=>true,
             'rsv1'=>false,
@@ -84,7 +84,7 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
 
         $msg = $codec->decode("\x81\x88\x00\xFF\x00\xFFa\x9dc\x9be\x99g\x97");
         $this->assertEquals([
-            'opcode'=>WebSocketCodec::OP_FRAME_TEXT,
+            'opcode'=>WebSocketConnection::OP_FRAME_TEXT,
             'payload'=>"abcdefgh",
             'final'=>true,
             'rsv1'=>false,