diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cc92ff..ef5fe7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ Blax Software's backward-compatibility guarantee. ### Added +- **Working browser-to-browser (P2P) MVP.** `Signaling\RelaySignalingHandler` (a + `blax-software/laravel-ws` `MessageHandler`) relays SDP/ICE between browsers so + they connect peer-to-peer — no server-side media engine required. Protocol: + `welcome` on connect, `join` → `joined` + `peer-joined`, targeted `relay`, room + `broadcast`, `leave`/disconnect → `peer-left`. Peers tracked in the kernel + `ConnectionRegistry`, membership in `RoomManager`. +- `webrtc:serve` now runs this relay over a `laravel-ws` WebSocketServer, so it is + browser-usable out of the box (`ws://host:port`). +- `require blax-software/laravel-ws`. +- Tests grew to 26 / 60 assertions, including a real WebSocket `welcome` + round-trip (browser → laravel-ws → relay). + +### Added (earlier) + - `WebRtcServiceProvider` + `config/webrtc.php` + the `webrtc:serve` command (boots signaling on the shared `blax-software/reactphp-kernel`). - `Contracts\MediaEngine` — the pluggable media backend diff --git a/README.md b/README.md index 6f54256..ca88b3c 100644 --- a/README.md +++ b/README.md @@ -5,21 +5,20 @@ [![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-17%20passing-success?style=flat-square)](#testing) -[![Assertions](https://img.shields.io/badge/assertions-42-blue?style=flat-square)](#testing) +[![Tests](https://img.shields.io/badge/tests-26%20passing-success?style=flat-square)](#testing) +[![Assertions](https://img.shields.io/badge/assertions-60-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. +WebRTC for Laravel on the shared [`reactphp-kernel`](https://github.com/blax-software/reactphp-kernel) backbone. A **WebSocket signaling relay** connects browsers for **peer-to-peer calls today** (no server-side media needed); when you need the server *in* the media path — recording, AI bridging, an SFU — a pluggable **media engine** (a Rust `str0m` core via `ext-php-rs`) takes over. ## Features -- 🎙️ **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** — `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/` -- 🎛️ **Transport-agnostic signaling** — route browser frames from this package's raw server or from `laravel-websockets` into the same handler +- 📞 **Browser-to-browser calls, working today** — two browsers `join` a room and the server relays their SDP/ICE so they connect **peer-to-peer**; the audio never touches the server, so no media engine is required +- 👥 **Rooms & mesh signaling** — peer discovery on join, `peer-joined` / `peer-left` notifications, targeted `relay`, and room `broadcast` +- 🧩 **On the shared kernel** — runs on `reactphp-kernel` over the [`laravel-ws`](https://github.com/blax-software/laravel-ws) WebSocket transport; one process, one loop, alongside your other realtime servers +- 🎙️ **Server-side media, when you need it** — a pluggable `MediaEngine` terminates media for **recording / intercept**, **AI-realtime bridging**, and **SFU** group calls +- 🤖 **AI realtime, provider hidden** — bridge a caller to an external model (e.g. OpenAI Realtime) server-side; the browser only ever talks to *you*, and the model is a config swap +- 🔌 **Pluggable engine** — `NullMediaEngine` (signaling-only) now; `Str0mMediaEngine` (Rust `str0m` via `ext-php-rs`, shipped in `rust/`) for real server-terminated media ## Installation @@ -28,36 +27,64 @@ composer require blax-software/laravel-webrtc php artisan vendor:publish --tag="webrtc-config" ``` -It depends on [`blax-software/reactphp-kernel`](https://github.com/blax-software/reactphp-kernel), resolved from git.blax.at via the `repositories` entry in `composer.json`. +It depends on [`blax-software/reactphp-kernel`](https://github.com/blax-software/reactphp-kernel) + [`blax-software/laravel-ws`](https://github.com/blax-software/laravel-ws), resolved from git.blax.at via the `repositories` entry in `composer.json`. ## Quick Start -Run the signaling server (on the shared kernel): +Run the signaling relay (browsers connect here): ```bash -php artisan webrtc:serve +php artisan webrtc:serve # ws://127.0.0.1:8090 by default ``` -With the default `NullMediaEngine`, signaling runs and the engine refuses `offer()` loudly. Build the Rust core (see [`rust/README.md`](rust/README.md)) and switch the engine for real calls: +That's all a **peer-to-peer call** needs — the server only relays signaling; the browsers exchange audio directly: + +```js +const ws = new WebSocket('ws://127.0.0.1:8090') +const pc = new RTCPeerConnection() +let peer + +ws.onmessage = async ({ data }) => { + const m = JSON.parse(data) + if (m.type === 'welcome') ws.send(JSON.stringify({ type: 'join', room: 'lobby' })) + if (m.type === 'joined') m.peers.forEach(p => (peer = p, call(p))) // someone's already here → call them + if (m.type === 'peer-joined') peer = m.peer // they'll send us an offer + if (m.type === 'relay') { // SDP / ICE from the other browser + peer = m.from + if (m.data.sdp) { await pc.setRemoteDescription(m.data); if (m.data.type === 'offer') answer() } + if (m.data.candidate) await pc.addIceCandidate(m.data) + } +} +pc.onicecandidate = e => e.candidate && ws.send(JSON.stringify({ type: 'relay', to: peer, data: e.candidate })) +pc.ontrack = e => audioEl.srcObject = e.streams[0] + +const send = (data) => ws.send(JSON.stringify({ type: 'relay', to: peer, data })) +async function call() { const o = await pc.createOffer(); await pc.setLocalDescription(o); send(o) } +async function answer() { const a = await pc.createAnswer(); await pc.setLocalDescription(a); send(a) } +``` + +### Relay protocol (JSON over WebSocket) + +```jsonc +← { "type": "welcome", "peer": "" } // on connect +→ { "type": "join", "room": "lobby" } +← { "type": "joined", "room": "lobby", "peers": ["", ...] } // who's already here +← { "type": "peer-joined", "room": "lobby", "peer": "" } // sent to the others +→ { "type": "relay", "to": "", "data": { ...sdp | candidate... } } +← { "type": "relay", "from": "", "data": { ... } } // delivered to the target +→ { "type": "broadcast", "data": { ... } } // to the rest of your room +→ { "type": "leave" } ← { "type": "left" } / others get { "type": "peer-left", "peer" } +``` + +### Server-terminated media (recording / AI bridge / SFU) + +When the server must be *in* the media path, configure a `MediaEngine`. `NullMediaEngine` (default) is signaling-only; `Str0mMediaEngine` is backed by a Rust `str0m` core via `ext-php-rs` (see [`rust/README.md`](rust/README.md)): ```dotenv WEBRTC_MEDIA_ENGINE="Blax\WebRtc\Media\Str0mMediaEngine" ``` -### Signaling wire shape (newline-delimited JSON) - -```jsonc -{ "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" } // 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" } // 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. +The engine-backed path uses `Signaling\SignalingHandler` (offer/ice/record/connect/bridge/close) + `Server\SignalingServer`; it's independent of the browser relay above and shares the same `RoomManager`. ## Configuration @@ -65,15 +92,17 @@ WEBRTC_MEDIA_ENGINE="Blax\WebRtc\Media\Str0mMediaEngine" ## Status -**MVP scaffold.** Signaling + the engine abstraction are real, tested, and runnable with `NullMediaEngine`; the `Str0mMediaEngine` + `rust/` core are the seam to build next. This is the WebRTC half of the blax realtime stack. +**Working MVP.** Browser-to-browser (mesh) calls work today via the WebSocket signaling relay — no server-side media required. Rooms, peer discovery, targeted relay and broadcast are tested, including a real WebSocket round-trip. Server-terminated media (recording, AI bridge, SFU) is the pluggable `MediaEngine`; the real one (`Str0mMediaEngine` + `rust/`) is the next build. ## Architecture ``` -reactphp-kernel shared backbone (loop, IPC, signals) - └─ laravel-webrtc (this) signaling on the kernel - ├─ NullMediaEngine signaling-only (dev/tests) - └─ Str0mMediaEngine Rust/str0m core via ext-php-rs (rust/) +reactphp-kernel shared backbone (loop, IPC, signals) + ├─ laravel-ws WebSocket transport (RFC6455) + └─ laravel-webrtc (this) + ├─ RelaySignalingHandler browser P2P calls (no media engine needed) + └─ MediaEngine (optional) server-terminated media + └─ Str0mMediaEngine Rust/str0m via ext-php-rs (rust/) ``` ## Testing @@ -83,7 +112,7 @@ composer install composer test ``` -The signaling router and engines are unit-tested against a fake media engine — no browser, sockets, or external services required. +The relay + signaling routers and engines are unit-tested with fakes; a real WebSocket round-trip proves the relay over `laravel-ws`, and a raw-socket test drives the engine signaling server — no browser or external services required. ## License diff --git a/composer.json b/composer.json index ea4ecc0..ef26566 100644 --- a/composer.json +++ b/composer.json @@ -12,6 +12,7 @@ ], "require": { "php": "^8.1|^8.2|^8.3|^8.4", + "blax-software/laravel-ws": "dev-master", "blax-software/reactphp-kernel": "dev-master", "react/event-loop": "^1.5", "react/socket": "^1.15", @@ -23,13 +24,16 @@ "laravel/pint": "^1.13" }, "suggest": { - "ext-ffi": "Required by the Str0m media engine (Rust core exposed to PHP).", - "blax-software/laravel-websockets": "Ride its WebSocket transport for browser signaling instead of the built-in raw signaling server." + "ext-ffi": "Required by the Str0m media engine (Rust core exposed to PHP)." }, "repositories": [ { "type": "vcs", "url": "ssh://git@git.blax.at:222/blax-software/reactphp-kernel.git" + }, + { + "type": "vcs", + "url": "ssh://git@git.blax.at:222/blax-software/laravel-ws.git" } ], "autoload": { diff --git a/src/Console/Commands/StartWebRtcServer.php b/src/Console/Commands/StartWebRtcServer.php index 048a360..8853d13 100644 --- a/src/Console/Commands/StartWebRtcServer.php +++ b/src/Console/Commands/StartWebRtcServer.php @@ -5,39 +5,30 @@ declare(strict_types=1); namespace Blax\WebRtc\Console\Commands; use Blax\ReactPhpKernel\Kernel; -use Blax\WebRtc\Contracts\MediaEngine; -use Blax\WebRtc\Media\NullMediaEngine; -use Blax\WebRtc\Server\SignalingServer; -use Blax\WebRtc\Signaling\SignalingHandler; +use Blax\WebRtc\Signaling\RelaySignalingHandler; +use Blax\Ws\WebSocketServer; use Illuminate\Console\Command; class StartWebRtcServer extends Command { protected $signature = 'webrtc:serve {--host= : Bind host (defaults to config webrtc.host)} {--port= : Bind port (defaults to config webrtc.port)}'; - protected $description = 'Start the blax WebRTC signaling server on the shared ReactPHP kernel.'; + protected $description = 'Start the blax WebRTC signaling relay (WebSocket) on the shared ReactPHP kernel — browsers connect here for peer-to-peer calls.'; public function handle(): int { $host = (string) ($this->option('host') ?: config('webrtc.host', '127.0.0.1')); $port = (int) ($this->option('port') ?: config('webrtc.port', 8090)); - $engine = app(config('webrtc.media_engine', NullMediaEngine::class)); - if (! $engine instanceof MediaEngine) { - $this->error('Configured webrtc.media_engine is not a '.MediaEngine::class); - - return self::FAILURE; - } - $kernel = (new Kernel)->reapChildren(); - $kernel->register(new SignalingServer( + $kernel->register(new WebSocketServer( $host, $port, - new SignalingHandler($engine), + new RelaySignalingHandler, (array) config('webrtc.tls', []), )); - $this->info("blax WebRTC signaling on {$host}:{$port} (engine: {$engine->name()}) — Ctrl-C to stop"); + $this->info("blax WebRTC signaling relay (WebSocket) on ws://{$host}:{$port} — browsers connect here for P2P calls. Ctrl-C to stop."); $kernel->run(); diff --git a/src/Signaling/RelaySignalingHandler.php b/src/Signaling/RelaySignalingHandler.php new file mode 100644 index 0000000..d847309 --- /dev/null +++ b/src/Signaling/RelaySignalingHandler.php @@ -0,0 +1,156 @@ +" } (on connect) + * → { "type": "join", "room": "lobby" } + * ← { "type": "joined", "room": "lobby", "peers": [...] } (existing peers) + * ← { "type": "peer-joined", "room": "lobby", "peer": "" } (to the others) + * → { "type": "relay", "to": "", "data": { ...sdp/ice... } } + * ← { "type": "relay", "from": "", "data": { ... } } (to the target) + * → { "type": "broadcast", "data": { ... } } (to the rest of your room) + * → { "type": "leave" } ← { "type": "left" } / peers get { "type": "peer-left", "peer" } + */ +final class RelaySignalingHandler implements MessageHandler +{ + private ConnectionRegistry $peers; + + private RoomManager $rooms; + + public function __construct(?ConnectionRegistry $peers = null, ?RoomManager $rooms = null) + { + $this->peers = $peers ?? new ConnectionRegistry; + $this->rooms = $rooms ?? new RoomManager; + } + + public function peers(): ConnectionRegistry + { + return $this->peers; + } + + public function rooms(): RoomManager + { + return $this->rooms; + } + + public function onOpen(Sendable $connection): void + { + $this->peers->add($connection->id(), $connection); + $connection->send($this->encode(['type' => 'welcome', 'peer' => $connection->id()])); + } + + public function onMessage(Sendable $connection, string $payload): void + { + $message = json_decode($payload, true); + if (! is_array($message)) { + $connection->send($this->error('invalid json')); + + return; + } + + $peer = $connection->id(); + + switch ((string) ($message['type'] ?? '')) { + case 'join': + $room = (string) ($message['room'] ?? ''); + if ($room === '') { + $connection->send($this->error('missing room')); + + return; + } + $others = $this->rooms->join($room, $peer); + $connection->send($this->encode(['type' => 'joined', 'room' => $room, 'peers' => $others])); + foreach ($others as $other) { + $this->sendTo($other, ['type' => 'peer-joined', 'room' => $room, 'peer' => $peer]); + } + break; + + case 'leave': + foreach ($this->rooms->leave($peer) as $other) { + $this->sendTo($other, ['type' => 'peer-left', 'peer' => $peer]); + } + $connection->send($this->encode(['type' => 'left'])); + break; + + case 'relay': + $to = (string) ($message['to'] ?? ''); + if ($to === '') { + $connection->send($this->error('missing to')); + + return; + } + if (! $this->sendTo($to, ['type' => 'relay', 'from' => $peer, 'data' => $message['data'] ?? null])) { + $connection->send($this->error("peer not connected: {$to}")); + } + break; + + case 'broadcast': + $room = $this->rooms->roomOf($peer); + if ($room !== null) { + foreach ($this->rooms->peersIn($room) as $other) { + if ($other !== $peer) { + $this->sendTo($other, ['type' => 'broadcast', 'from' => $peer, 'data' => $message['data'] ?? null]); + } + } + } + break; + + default: + $connection->send($this->error('unknown type')); + } + } + + public function onClose(Sendable $connection): void + { + $peer = $connection->id(); + foreach ($this->rooms->leave($peer) as $other) { + $this->sendTo($other, ['type' => 'peer-left', 'peer' => $peer]); + } + $this->peers->remove($peer); + } + + /** Deliver a message to a specific peer; false if they're not connected. */ + private function sendTo(string $peerId, array $message): bool + { + $connection = $this->peers->get($peerId); + if (! $connection instanceof Sendable) { + return false; + } + + $connection->send($this->encode($message)); + + return true; + } + + private function error(string $error): string + { + return $this->encode(['type' => 'error', 'error' => $error]); + } + + /** @param array $message */ + private function encode(array $message): string + { + return (string) json_encode($message); + } +} diff --git a/tests/Fixtures/FakePeer.php b/tests/Fixtures/FakePeer.php new file mode 100644 index 0000000..affcea0 --- /dev/null +++ b/tests/Fixtures/FakePeer.php @@ -0,0 +1,47 @@ + */ + public array $sent = []; + + public bool $closed = false; + + public function __construct(private string $id) {} + + public function id(): string + { + return $this->id; + } + + public function send(string $message): void + { + $this->sent[] = $message; + } + + public function close(): void + { + $this->closed = true; + } + + /** @return array> */ + public function jsonMessages(): array + { + return array_map(fn ($s) => json_decode($s, true) ?? [], $this->sent); + } + + /** @return array */ + public function lastJson(): array + { + $last = end($this->sent); + + return $last === false ? [] : (json_decode($last, true) ?? []); + } +} diff --git a/tests/RelaySignalingHandlerTest.php b/tests/RelaySignalingHandlerTest.php new file mode 100644 index 0000000..53fad42 --- /dev/null +++ b/tests/RelaySignalingHandlerTest.php @@ -0,0 +1,132 @@ +onOpen($a); + + $this->assertSame(['type' => 'welcome', 'peer' => 'a'], $a->lastJson()); + } + + public function test_join_returns_existing_peers_and_notifies_them(): void + { + $handler = new RelaySignalingHandler; + $a = new FakePeer('a'); + $b = new FakePeer('b'); + $handler->onOpen($a); + $handler->onOpen($b); + + $handler->onMessage($a, $this->json(['type' => 'join', 'room' => 'lobby'])); + $handler->onMessage($b, $this->json(['type' => 'join', 'room' => 'lobby'])); + + // b learns 'a' is already there + $this->assertSame(['type' => 'joined', 'room' => 'lobby', 'peers' => ['a']], $b->lastJson()); + // a is told b joined + $this->assertContains(['type' => 'peer-joined', 'room' => 'lobby', 'peer' => 'b'], $a->jsonMessages()); + } + + public function test_relay_forwards_the_payload_to_the_target_with_the_sender(): void + { + $handler = new RelaySignalingHandler; + $a = new FakePeer('a'); + $b = new FakePeer('b'); + $handler->onOpen($a); + $handler->onOpen($b); + + $handler->onMessage($a, $this->json(['type' => 'relay', 'to' => 'b', 'data' => ['sdp' => 'OFFER']])); + + $this->assertSame(['type' => 'relay', 'from' => 'a', 'data' => ['sdp' => 'OFFER']], $b->lastJson()); + } + + public function test_relay_to_an_unknown_peer_errors_the_sender(): void + { + $handler = new RelaySignalingHandler; + $a = new FakePeer('a'); + $handler->onOpen($a); + + $handler->onMessage($a, $this->json(['type' => 'relay', 'to' => 'ghost', 'data' => []])); + + $this->assertSame('error', $a->lastJson()['type']); + } + + public function test_leave_notifies_the_remaining_peers(): void + { + $handler = new RelaySignalingHandler; + $a = new FakePeer('a'); + $b = new FakePeer('b'); + $handler->onOpen($a); + $handler->onOpen($b); + $handler->onMessage($a, $this->json(['type' => 'join', 'room' => 'lobby'])); + $handler->onMessage($b, $this->json(['type' => 'join', 'room' => 'lobby'])); + + $handler->onMessage($b, $this->json(['type' => 'leave'])); + + $this->assertContains(['type' => 'peer-left', 'peer' => 'b'], $a->jsonMessages()); + $this->assertSame(['type' => 'left'], $b->lastJson()); + } + + public function test_disconnect_leaves_the_room_and_forgets_the_peer(): void + { + $handler = new RelaySignalingHandler; + $a = new FakePeer('a'); + $b = new FakePeer('b'); + $handler->onOpen($a); + $handler->onOpen($b); + $handler->onMessage($a, $this->json(['type' => 'join', 'room' => 'lobby'])); + $handler->onMessage($b, $this->json(['type' => 'join', 'room' => 'lobby'])); + + $handler->onClose($b); + + $this->assertContains(['type' => 'peer-left', 'peer' => 'b'], $a->jsonMessages()); + $this->assertFalse($handler->peers()->has('b')); + } + + public function test_broadcast_reaches_the_rest_of_the_room_only(): void + { + $handler = new RelaySignalingHandler; + $peers = [new FakePeer('a'), new FakePeer('b'), new FakePeer('c')]; + foreach ($peers as $p) { + $handler->onOpen($p); + $handler->onMessage($p, $this->json(['type' => 'join', 'room' => 'r'])); + } + [$a, $b, $c] = $peers; + + $handler->onMessage($a, $this->json(['type' => 'broadcast', 'data' => ['x' => 1]])); + + $expected = ['type' => 'broadcast', 'from' => 'a', 'data' => ['x' => 1]]; + $this->assertContains($expected, $b->jsonMessages()); + $this->assertContains($expected, $c->jsonMessages()); + $this->assertNotContains($expected, $a->jsonMessages()); + } + + public function test_invalid_json_and_unknown_type_are_errors(): void + { + $handler = new RelaySignalingHandler; + $a = new FakePeer('a'); + $handler->onOpen($a); + + $handler->onMessage($a, 'not json'); + $this->assertSame('error', $a->lastJson()['type']); + + $handler->onMessage($a, $this->json(['type' => 'wat'])); + $this->assertSame('error', $a->lastJson()['type']); + } + + /** @param array $data */ + private function json(array $data): string + { + return (string) json_encode($data); + } +} diff --git a/tests/WebSocketRelayServerTest.php b/tests/WebSocketRelayServerTest.php new file mode 100644 index 0000000..9881f7f --- /dev/null +++ b/tests/WebSocketRelayServerTest.php @@ -0,0 +1,101 @@ +boot($loop); + + $address = $server->address(); + $this->assertNotNull($address); + $port = (int) parse_url($address, PHP_URL_PORT); + + $handshakeOk = false; + $welcome = null; + + (new Connector([], $loop)) + ->connect("127.0.0.1:{$port}") + ->then(function (ConnectionInterface $conn) use (&$handshakeOk, &$welcome, $loop) { + $buffer = ''; + $upgraded = false; + + $conn->on('data', function ($data) use (&$buffer, &$upgraded, &$handshakeOk, &$welcome, $conn, $loop) { + $buffer .= $data; + + if (! $upgraded) { + $end = strpos($buffer, "\r\n\r\n"); + if ($end === false) { + return; + } + $handshakeOk = str_contains(substr($buffer, 0, $end), ' 101 '); + $buffer = substr($buffer, $end + 4); + $upgraded = true; + if (! $handshakeOk) { + $loop->stop(); + + return; + } + } + + // The server pushes a `welcome` frame on connect (no client frame needed). + $decoded = self::decodeServerTextFrame($buffer); + if ($decoded !== null) { + $welcome = json_decode($decoded, true); + $conn->close(); + $loop->stop(); + } + }); + + $key = base64_encode(random_bytes(16)); + $conn->write( + "GET / HTTP/1.1\r\n". + "Host: 127.0.0.1\r\n". + "Upgrade: websocket\r\n". + "Connection: Upgrade\r\n". + "Sec-WebSocket-Key: {$key}\r\n". + "Sec-WebSocket-Version: 13\r\n\r\n" + ); + }); + + $loop->addTimer(3.0, fn () => $loop->stop()); + $loop->run(); + $server->shutdown(); + + $this->assertTrue($handshakeOk, 'server should return a 101 upgrade'); + $this->assertSame('welcome', $welcome['type'] ?? null); + $this->assertArrayHasKey('peer', (array) $welcome); + } + + /** Decode one unmasked server->client text frame (payload < 126 bytes), or null. */ + private static function decodeServerTextFrame(string $data): ?string + { + if (strlen($data) < 2) { + return null; + } + $length = ord($data[1]) & 0x7F; + if ($length >= 126 || strlen($data) < 2 + $length) { + return null; + } + + return substr($data, 2, $length); + } +}