From 1479e36f494d4df3233098c7cf6552b90e6d9de6 Mon Sep 17 00:00:00 2001 From: Blax Software Date: Tue, 7 Jul 2026 13:54:36 +0200 Subject: [PATCH] feat: multi-party rooms + real socket round-trip test - Rooms\Room + Rooms\RoomManager: track peers per room, auto-mesh a new joiner with everyone already there, drop empty rooms - SignalingHandler: join/leave message types (auto-connectPeers on join), and close now also leaves the room; expose rooms() - SignalingServer::address() for the bound listen address - Tests: real offer->answer round-trip over a live TCP socket on the kernel loop, RoomManager + join/leave/close coverage. Suite: 17 tests / 42 assertions - Verified on the updated reactphp-kernel (logging + registry) rel learn-atc #1055 #1051 #1019 Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 7 +++ README.md | 12 +++-- src/Rooms/Room.php | 57 ++++++++++++++++++++ src/Rooms/RoomManager.php | 86 ++++++++++++++++++++++++++++++ src/Server/SignalingServer.php | 6 +++ src/Signaling/SignalingHandler.php | 60 +++++++++++++++++++-- tests/RoomManagerTest.php | 52 ++++++++++++++++++ tests/SignalingHandlerTest.php | 47 ++++++++++++++++ tests/SignalingServerTest.php | 55 +++++++++++++++++++ 9 files changed, 374 insertions(+), 8 deletions(-) create mode 100644 src/Rooms/Room.php create mode 100644 src/Rooms/RoomManager.php create mode 100644 tests/RoomManagerTest.php create mode 100644 tests/SignalingServerTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a8243d..0cc92ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,3 +22,10 @@ Blax Software's backward-compatibility guarantee. handler. - `rust/` — the `blax_webrtc` crate scaffold (str0m + ext-php-rs) documenting the media plan. +- `Rooms\Room` + `Rooms\RoomManager` and `join`/`leave` signaling for + party-to-party / group calls: a new joiner is auto-meshed with everyone already + in the room; `close` also leaves. Empty rooms are dropped. +- `SignalingServer::address()` — the bound listen address (for integration). +- A real socket round-trip test (offer→answer over a live TCP connection on the + kernel loop), plus RoomManager + join/leave coverage. Suite: 17 tests / 42 + assertions. diff --git a/README.md b/README.md index 045e2dc..6f54256 100644 --- a/README.md +++ b/README.md @@ -5,8 +5,8 @@ [![PHP Version](https://img.shields.io/badge/php-%5E8.1-blue?style=flat-square)](https://php.net) [![Laravel](https://img.shields.io/badge/laravel-10.x--12.x-orange?style=flat-square)](https://laravel.com) [![Built on](https://img.shields.io/badge/on-reactphp--kernel-4b275f?style=flat-square)](https://github.com/blax-software/reactphp-kernel) -[![Tests](https://img.shields.io/badge/tests-8%20passing-success?style=flat-square)](#testing) -[![Assertions](https://img.shields.io/badge/assertions-18-blue?style=flat-square)](#testing) +[![Tests](https://img.shields.io/badge/tests-17%20passing-success?style=flat-square)](#testing) +[![Assertions](https://img.shields.io/badge/assertions-42-blue?style=flat-square)](#testing) [![License](https://img.shields.io/badge/license-MIT-green?style=flat-square)](LICENSE) WebRTC for Laravel on the shared [`reactphp-kernel`](https://github.com/blax-software/reactphp-kernel) backbone. PHP owns **signaling + orchestration**; a pluggable **media engine** owns the real-time media (ICE/DTLS/SRTP/RTP/Opus) — with the production engine backed by a **Rust core (`str0m`) via `ext-php-rs`**, so PHP never runs per-20ms DSP. @@ -15,7 +15,7 @@ WebRTC for Laravel on the shared [`reactphp-kernel`](https://github.com/blax-sof - 🎙️ **Record / intercept / log** — the engine terminates media server-side, so every participant's audio (human or AI) can be captured - 🤖 **AI realtime, provider hidden** — bridge a caller to an external realtime model (e.g. OpenAI Realtime) over a server-side socket; the browser only ever talks to *you*, and the model is a config swap -- 👥 **Party-to-party & group calls** — instructor↔student, controller↔pilot, guest↔admin are just peers the engine routes between +- 👥 **Party-to-party & group calls** — `join`/`leave` a room and the `RoomManager` meshes each new peer with everyone already there (instructor↔student, controller↔pilot, guest↔admin) - 🧩 **On the shared kernel** — signaling runs on `reactphp-kernel` alongside your WebSockets, one process, one loop - 🔌 **Pluggable `MediaEngine`** — `NullMediaEngine` for signaling-only dev; `Str0mMediaEngine` for the real Rust-backed media - 🦀 **Rust media core** — `str0m` (sans-IO WebRTC) via `ext-php-rs`, shipped in `rust/` @@ -50,9 +50,11 @@ WEBRTC_MEDIA_ENGINE="Blax\WebRtc\Media\Str0mMediaEngine" { "type": "offer", "peer": "p1", "sdp": "..." } // -> { "type": "answer", "peer": "p1", "sdp": "..." } { "type": "ice", "peer": "p1", "candidate": {...} } // (no reply) { "type": "record", "peer": "p1", "path": "..." } // -> { "type": "ack", ... } -{ "type": "connect", "peer": "p1", "other": "p2" } // party-to-party +{ "type": "connect", "peer": "p1", "other": "p2" } // low-level: mesh two peers +{ "type": "join", "peer": "p1", "room": "lobby" } // -> { "type": "joined", "peers": [...] } (auto-meshes) +{ "type": "leave", "peer": "p1" } // -> { "type": "left", "peers": [...] } { "type": "bridge", "peer": "p1", "options": {...} } // dial an external realtime model -{ "type": "close", "peer": "p1" } +{ "type": "close", "peer": "p1" } // closes media + leaves the room ``` `SignalingHandler` is transport-agnostic — feed it frames from the built-in raw `SignalingServer` today, or from the `laravel-websockets` WS transport (for browsers) later, without changing the media wiring. diff --git a/src/Rooms/Room.php b/src/Rooms/Room.php new file mode 100644 index 0000000..c75df2f --- /dev/null +++ b/src/Rooms/Room.php @@ -0,0 +1,57 @@ + peer id => present */ + private array $peers = []; + + public function __construct(public readonly string $id) {} + + /** + * Add a peer and return the OTHER peers already present — the set the caller + * must mesh the new peer with (engine->connectPeers). + * + * @return array + */ + public function add(string $peerId): array + { + $others = array_keys($this->peers); + $this->peers[$peerId] = true; + + return $others; + } + + public function remove(string $peerId): void + { + unset($this->peers[$peerId]); + } + + public function has(string $peerId): bool + { + return isset($this->peers[$peerId]); + } + + /** @return array */ + public function peers(): array + { + return array_keys($this->peers); + } + + public function isEmpty(): bool + { + return $this->peers === []; + } + + public function count(): int + { + return count($this->peers); + } +} diff --git a/src/Rooms/RoomManager.php b/src/Rooms/RoomManager.php new file mode 100644 index 0000000..784dbd7 --- /dev/null +++ b/src/Rooms/RoomManager.php @@ -0,0 +1,86 @@ + roomId => Room */ + private array $rooms = []; + + /** @var array peerId => roomId */ + private array $peerRoom = []; + + /** + * Join $peerId to $roomId, returning the OTHER peers already present so the + * caller can mesh them. + * + * @return array + */ + public function join(string $roomId, string $peerId): array + { + // Moving between rooms: leave the old one first. + if (($this->peerRoom[$peerId] ?? null) !== null && $this->peerRoom[$peerId] !== $roomId) { + $this->leave($peerId); + } + + $room = $this->rooms[$roomId] ??= new Room($roomId); + $others = $room->add($peerId); + $this->peerRoom[$peerId] = $roomId; + + return $others; + } + + /** + * Remove $peerId from its room, returning the peers that REMAIN there (so the + * caller can notify them). Drops the room when it empties. + * + * @return array + */ + public function leave(string $peerId): array + { + $roomId = $this->peerRoom[$peerId] ?? null; + if ($roomId === null || ! isset($this->rooms[$roomId])) { + return []; + } + + $room = $this->rooms[$roomId]; + $room->remove($peerId); + unset($this->peerRoom[$peerId]); + + if ($room->isEmpty()) { + unset($this->rooms[$roomId]); + + return []; + } + + return $room->peers(); + } + + public function roomOf(string $peerId): ?string + { + return $this->peerRoom[$peerId] ?? null; + } + + public function room(string $roomId): ?Room + { + return $this->rooms[$roomId] ?? null; + } + + /** @return array */ + public function peersIn(string $roomId): array + { + return isset($this->rooms[$roomId]) ? $this->rooms[$roomId]->peers() : []; + } + + public function count(): int + { + return count($this->rooms); + } +} diff --git a/src/Server/SignalingServer.php b/src/Server/SignalingServer.php index 0e5417d..2990b15 100644 --- a/src/Server/SignalingServer.php +++ b/src/Server/SignalingServer.php @@ -40,6 +40,12 @@ final class SignalingServer implements Server return 'webrtc-signaling'; } + /** The bound listen address (e.g. "tcp://127.0.0.1:54321"), or null before boot. */ + public function address(): ?string + { + return $this->socket?->getAddress(); + } + public function boot(LoopInterface $loop): void { $this->socket = SocketServerFactory::create("{$this->host}:{$this->port}", $loop, $this->tls); diff --git a/src/Signaling/SignalingHandler.php b/src/Signaling/SignalingHandler.php index 56831a0..cd0edcc 100644 --- a/src/Signaling/SignalingHandler.php +++ b/src/Signaling/SignalingHandler.php @@ -5,6 +5,7 @@ declare(strict_types=1); namespace Blax\WebRtc\Signaling; use Blax\WebRtc\Contracts\MediaEngine; +use Blax\WebRtc\Rooms\RoomManager; /** * Transport-agnostic signaling router: decodes a signaling message and drives @@ -15,11 +16,22 @@ use Blax\WebRtc\Contracts\MediaEngine; * the browser-facing transport (this package's raw SignalingServer today, or the * laravel-websockets WS layer later) is a swappable detail. * - * Wire shape (JSON): { "type": "offer|ice|record|connect|bridge|close", "peer": "...", ... } + * Wire shape (JSON): { "type": "offer|ice|record|connect|bridge|close|join|leave", "peer": "...", ... } */ final class SignalingHandler { - public function __construct(private MediaEngine $engine) {} + private RoomManager $rooms; + + public function __construct(private MediaEngine $engine, ?RoomManager $rooms = null) + { + $this->rooms = $rooms ?? new RoomManager; + } + + /** The shared room registry (party-to-party / group calls). */ + public function rooms(): RoomManager + { + return $this->rooms; + } /** * @param array $message @@ -45,7 +57,9 @@ final class SignalingHandler 'record' => $this->ack($peer, fn () => $this->engine->startRecording($peer, (string) ($message['path'] ?? ''))), 'connect' => $this->ack($peer, fn () => $this->engine->connectPeers($peer, (string) ($message['other'] ?? ''))), 'bridge' => $this->ack($peer, fn () => $this->engine->bridge($peer, (array) ($message['options'] ?? []))), - 'close' => $this->ack($peer, fn () => $this->engine->close($peer)), + 'close' => $this->close($peer), + 'join' => $this->join($peer, (string) ($message['room'] ?? '')), + 'leave' => $this->leave($peer), default => ['type' => 'error', 'error' => "unknown type: {$type}"], }; } catch (\Throwable $e) { @@ -68,4 +82,44 @@ final class SignalingHandler return ['type' => 'ack', 'peer' => $peer]; } + + /** Tear a peer down: drop it from its room, then close its media. */ + private function close(string $peer): array + { + $this->rooms->leave($peer); + $this->engine->close($peer); + + return ['type' => 'ack', 'peer' => $peer]; + } + + /** + * Join a room and mesh with everyone already there. + * + * @return array + */ + private function join(string $peer, string $roomId): array + { + if ($roomId === '') { + return ['type' => 'error', 'peer' => $peer, 'error' => 'missing room']; + } + + $others = $this->rooms->join($roomId, $peer); + foreach ($others as $other) { + $this->engine->connectPeers($peer, $other); + } + + return ['type' => 'joined', 'peer' => $peer, 'room' => $roomId, 'peers' => $others]; + } + + /** + * Leave the current room; the reply lists the peers that remain. + * + * @return array + */ + private function leave(string $peer): array + { + $remaining = $this->rooms->leave($peer); + + return ['type' => 'left', 'peer' => $peer, 'peers' => $remaining]; + } } diff --git a/tests/RoomManagerTest.php b/tests/RoomManagerTest.php new file mode 100644 index 0000000..8eaf22e --- /dev/null +++ b/tests/RoomManagerTest.php @@ -0,0 +1,52 @@ +assertSame([], $rooms->join('lobby', 'p1')); + $this->assertSame(['p1'], $rooms->peersIn('lobby')); + $this->assertSame('lobby', $rooms->roomOf('p1')); + } + + public function test_second_joiner_meshes_with_the_first(): void + { + $rooms = new RoomManager; + $rooms->join('lobby', 'p1'); + + $this->assertSame(['p1'], $rooms->join('lobby', 'p2')); + $this->assertSame(['p1', 'p2'], $rooms->peersIn('lobby')); + } + + public function test_leaving_returns_remaining_and_drops_empty_rooms(): void + { + $rooms = new RoomManager; + $rooms->join('lobby', 'p1'); + $rooms->join('lobby', 'p2'); + + $this->assertSame(['p1'], $rooms->leave('p2')); + $this->assertSame([], $rooms->leave('p1')); // room now empty + $this->assertSame(0, $rooms->count()); + $this->assertNull($rooms->room('lobby')); + } + + public function test_joining_a_new_room_leaves_the_old(): void + { + $rooms = new RoomManager; + $rooms->join('a', 'p1'); + $rooms->join('b', 'p1'); + + $this->assertSame([], $rooms->peersIn('a')); // 'a' emptied + dropped + $this->assertSame(['p1'], $rooms->peersIn('b')); + $this->assertSame('b', $rooms->roomOf('p1')); + } +} diff --git a/tests/SignalingHandlerTest.php b/tests/SignalingHandlerTest.php index a36d87a..f657dd8 100644 --- a/tests/SignalingHandlerTest.php +++ b/tests/SignalingHandlerTest.php @@ -76,4 +76,51 @@ final class SignalingHandlerTest extends TestCase $this->assertSame(['type' => 'error', 'peer' => 'p1', 'error' => 'boom'], $result); } + + public function test_join_meshes_a_new_peer_with_those_already_present(): void + { + $engine = new FakeMediaEngine; + $handler = new SignalingHandler($engine); + + $this->assertSame( + ['type' => 'joined', 'peer' => 'p1', 'room' => 'lobby', 'peers' => []], + $handler->handle(['type' => 'join', 'peer' => 'p1', 'room' => 'lobby']), + ); + $this->assertSame( + ['type' => 'joined', 'peer' => 'p2', 'room' => 'lobby', 'peers' => ['p1']], + $handler->handle(['type' => 'join', 'peer' => 'p2', 'room' => 'lobby']), + ); + + $this->assertContains(['connect', 'p2', 'p1'], $engine->calls); + } + + public function test_join_without_a_room_is_an_error(): void + { + $result = (new SignalingHandler(new FakeMediaEngine))->handle(['type' => 'join', 'peer' => 'p1']); + + $this->assertSame('error', $result['type']); + } + + public function test_leave_reports_the_remaining_peers(): void + { + $handler = new SignalingHandler(new FakeMediaEngine); + $handler->handle(['type' => 'join', 'peer' => 'p1', 'room' => 'lobby']); + $handler->handle(['type' => 'join', 'peer' => 'p2', 'room' => 'lobby']); + + $this->assertSame( + ['type' => 'left', 'peer' => 'p2', 'peers' => ['p1']], + $handler->handle(['type' => 'leave', 'peer' => 'p2']), + ); + } + + public function test_close_also_removes_the_peer_from_its_room(): void + { + $engine = new FakeMediaEngine; + $handler = new SignalingHandler($engine); + $handler->handle(['type' => 'join', 'peer' => 'p1', 'room' => 'lobby']); + + $this->assertSame(['type' => 'ack', 'peer' => 'p1'], $handler->handle(['type' => 'close', 'peer' => 'p1'])); + $this->assertNull($handler->rooms()->roomOf('p1')); + $this->assertContains(['close', 'p1'], $engine->calls); + } } diff --git a/tests/SignalingServerTest.php b/tests/SignalingServerTest.php new file mode 100644 index 0000000..e9ce09f --- /dev/null +++ b/tests/SignalingServerTest.php @@ -0,0 +1,55 @@ +boot($loop); + + $address = $server->address(); + $this->assertNotNull($address, 'server should be bound after boot'); + $port = (int) parse_url($address, PHP_URL_PORT); + $this->assertGreaterThan(0, $port); + + $received = null; + + (new Connector([], $loop)) + ->connect("127.0.0.1:{$port}") + ->then(function (ConnectionInterface $conn) use (&$received, $loop) { + $buffer = ''; + $conn->on('data', function ($chunk) use (&$buffer, &$received, $conn, $loop) { + $buffer .= $chunk; + if (($pos = strpos($buffer, "\n")) !== false) { + $received = json_decode(substr($buffer, 0, $pos), true); + $conn->close(); + $loop->stop(); + } + }); + $conn->write(json_encode(['type' => 'offer', 'peer' => 'p1', 'sdp' => 'OFFER-SDP'])."\n"); + }); + + // Safety net so a broken server can't hang the suite. + $loop->addTimer(2.0, fn () => $loop->stop()); + $loop->run(); + $server->shutdown(); + + $this->assertSame(['type' => 'answer', 'peer' => 'p1', 'sdp' => 'answer-for-p1'], $received); + $this->assertSame(['offer', 'p1', 'OFFER-SDP'], $engine->calls[0]); + } +}