More inline docs

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,28 @@ 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);
+```
+
+## 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,9 @@ class ConnectionGroup implements EventEmitterInterface, IteratorAggregate, Count
     {
         return $this->name;
     }
+
+    public function write(string $payload)
+    {
+        
+    }
 }

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_CREATED = 'created';
+    /**
+     * @var string emitted after the last member leaves, when the group is destroyed
+     */
+    const EVENT_DESTROYED = 'destroyed';
+
     /** @var array<string,ConnectionGroup> */
     private array $groups = [];
 
@@ -20,11 +31,13 @@ class GroupManager
             $group->on(ConnectionGroup::EVENT_LEAVE, function () use ($group) {
                 Loop::futureTick(function () use ($group) {
                     if (count($group) === 0) {
+                        $this->emit(self::EVENT_DESTROYED, [ $group ]);
                         $group->removeAllListeners();
                         unset($this->groups[$group->getName()]);
                     }
                 });
             });
+            $this->emit(self::EVENT_CREATED, [ $group ]);
         } else {
             $group = $this->groups[$name];
         }

src/WebSocketCodec.php

@@ -17,19 +17,26 @@ use React\Stream\WritableStreamInterface;
 class WebSocketCodec
 {
 
-    const OP_CONTINUATION = 0x0;
+    /**
-    const OP_FRAME_TEXT = 0x1;
+     * Encode a frame. Required keys are opcode and payload. Keys that can be passed are:
-    const OP_FRAME_BINARY = 0x2;
+     * 
-    const OP_CLOSE = 0x8;
+     *   opcode (int) - the opcode
-    const OP_PING = 0x9;
+     *   final (bool) - final frame
-    const OP_PONG = 0xA;
+     *   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));
-        }
-    }
-
 }

src/WebSocketConnection.php

@@ -25,8 +25,10 @@ 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;
 
+    /** @var WebSocketCodec The frame encoder/decoder */
     private WebSocketCodec $codec;
 
     private ?ConnectionGroup $group = null;
@@ -54,6 +56,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 +93,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 +110,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 +173,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 +245,21 @@ class WebSocketConnection implements WebSocketInterface
         return true;
     }
 
+    /**
+     * {@inheritDoc}
+     */
     public function close()
     {
         $this->outStream->close();
         $this->inStream->close();
     }
 
+    /**
+     * {@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));
-    //     }
-    // }
-
 }

tests/WebSocketCodecTest.php

@@ -13,18 +13,18 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
         $codec = new WebSocketCodec();
 
         $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"
@@ -47,7 +47,7 @@ class WebSocketCodecTest extends \PHPUnit\Framework\TestCase
 
         $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,