Allow accessing websocket connections via protocol

src/main/php/xp/web/srv/WebSocketProtocol.class.php

@@ -22,6 +22,65 @@ public function __construct($listener, $logging= null) {
     $this->logging= $logging ?? new Logging(null);
   }
 
+  /** Adds a connection */
+  public function add(Connection $connection): Connection {
+    $this->connections[$connection->id()]= $connection;
+    return $connection;
+  }
+
+  /**
+   * Returns a given connection by its specified ID
+   *
+   * @param  int|string $id
+   * @return ?websocket.protocol.Connection
+   */
+  public function connection($id) {
+    return $this->connections[$id] ?? null;
+  }
+
+  /**
+   * Transmits a given payload to the specified targets, which may
+   * either be IDs or `websocket.protocol.Connection` instances.
+   *
+   * @param  iterable $targets
+   * @param  string|util.Bytes $payload
+   * @return void
+   */
+  public function transmit(iterable $targets, $payload) {
+    foreach ($targets as $target) {
+      if ($target instanceof Connection) {
+        $target->send($payload);
+      } else if ($connection= ($this->connections[$target] ?? null)) {
+        $connection->send($payload);
+      }
+    }
+  }
+
+  /**
+   * Broadcast a message to all connections
+   *
+   * @param  string|util.Bytes $payload
+   * @param  iterable $prioritize
+   * @return void
+   */
+  public function broadcast($payload, iterable $prioritize= []) {
+    $connections= $this->connections;
+
+    foreach ($prioritize as $target) {
+      if ($target instanceof Connection) {
+        $target->send($payload);
+        unset($connections[$connection->id()]);
+      } else if ($connection= ($this->connections[$target] ?? null)) {
+        $connection->send($payload);
+        unset($connections[$target]);
+      }
+    }
+
+    foreach ($connections as $connection) {
+      $connection->send($payload);
+    }
+  }
+
   /**
    * Handle client switch
    *

src/test/php/web/unittest/server/WebSocketProtocolTest.class.php

@@ -5,6 +5,7 @@
 use web\Logging;
 use web\unittest\Channel;
 use websocket\Listener;
+use websocket\protocol\Connection;
 use xp\web\srv\WebSocketProtocol;
 
 class WebSocketProtocolTest {
@@ -108,4 +109,66 @@ public function logs_messages($input, $expected) {
     });
     Assert::equals([$expected], $logged);
   }
+
+  #[Test]
+  public function without_connection() {
+    $fixture= new WebSocketProtocol($this->noop);
+
+    Assert::null($fixture->connection('test-1'));
+  }
+
+  #[Test]
+  public function add_connections() {
+    $fixture= new WebSocketProtocol($this->noop);
+    $a= $fixture->add(new Connection(null, 'test-a', '/'));
+    $b= $fixture->add(new Connection(null, 'test-b', '/'));
+
+    Assert::equals($a, $fixture->connection('test-a'));
+    Assert::equals($b, $fixture->connection('test-b'));
+  }
+
+  #[Test, Values([[['test-a'], ['a' => ['Hello']]], [['test-b'], ['b' => ['Hello']]]])]
+  public function transmit($targets, $expected) {
+    $transmitted= [];
+    $fixture= new WebSocketProtocol($this->noop);
+    $a= $fixture->add(newinstance(Connection::class, [null, 'test-a', '/'], [
+      'send' => function($payload) use(&$transmitted) { $transmitted['a'][]= $payload; }
+    ]));
+    $b= $fixture->add(newinstance(Connection::class, [null, 'test-b', '/'], [
+      'send' => function($payload) use(&$transmitted) { $transmitted['b'][]= $payload; }
+    ]));
+    $fixture->transmit($targets, 'Hello');
+
+    Assert::equals($expected, $transmitted);
+  }
+
+  #[Test]
+  public function broadcast() {
+    $transmitted= [];
+    $fixture= new WebSocketProtocol($this->noop);
+    $a= $fixture->add(newinstance(Connection::class, [null, 'test-a', '/'], [
+      'send' => function($payload) use(&$transmitted) { $transmitted['a'][]= $payload; }
+    ]));
+    $b= $fixture->add(newinstance(Connection::class, [null, 'test-b', '/'], [
+      'send' => function($payload) use(&$transmitted) { $transmitted['b'][]= $payload; }
+    ]));
+    $fixture->broadcast('Hello');
+
+    Assert::equals(['a' => ['Hello'], 'b' => ['Hello']], $transmitted);
+  }
+
+  #[Test, Values([[['test-a'], ['a', 'b']], [['test-b'], ['b', 'a']]])]
+  public function broadcast_prioritizing($targets, $order) {
+    $transmitted= [];
+    $fixture= new WebSocketProtocol($this->noop);
+    $a= $fixture->add(newinstance(Connection::class, [null, 'test-a', '/'], [
+      'send' => function($payload) use(&$transmitted) { $transmitted['a'][]= $payload; }
+    ]));
+    $b= $fixture->add(newinstance(Connection::class, [null, 'test-b', '/'], [
+      'send' => function($payload) use(&$transmitted) { $transmitted['b'][]= $payload; }
+    ]));
+    $fixture->broadcast('Hello', $targets);
+
+    Assert::equals($order, array_keys($transmitted));
+  }
 }