Bugfixed handling of fin flag in WebSocketConnection

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)
+```

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.

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

@@ -61,8 +61,20 @@ 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);
+            }
+        }
     }
+
+    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'];
@@ -252,6 +257,14 @@ class WebSocketConnection implements WebSocketInterface
     {
         $this->outStream->close();
         $this->inStream->close();
+        
+        $this->emit('close', []);        
+    }
+
+    public function closeWithReason(string $reason, int $code=1000)
+    {
+        $payload = chr(($code >> 8) & 0xFF) . chr($code & 0xFF) . $reason;
+        $this->send(self::OP_CLOSE, $payload);
     }
 
     /**
@@ -259,7 +272,7 @@ class WebSocketConnection implements WebSocketInterface
      */
     public function end($data = null)
     {
-        
+        // TODO implement me
     }
 
 }

src/WebSocketInterface.php

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

src/WebSocketProtocol.php

@@ -14,7 +14,7 @@ use React\Stream\ReadableStreamInterface;
 use React\Stream\ThroughStream;
 use React\Stream\WritableStreamInterface;
 
-class WebSocketCodec
+class WebSocketProtocol
 {
 
     /**

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
 {
 

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([