feat: transport-agnostic domain layer — run the relay on any WebSocket
Rooms / recording / realtime bridge no longer require the bundled ReactPHP server, so a host (e.g. a fork-per-message Laravel WS) can consume the package on its own transport: - RoomStore contract + ArrayRoomStore; RoomManager takes a store so membership can be Cache/Redis-backed and survive across workers - Realtime\OpenAiRealtimeBridge (+ RealtimeTransport / TextalkRealtimeTransport / RealtimeTurn): the OpenAI GA realtime WS mechanism, turn-based, unit-tested via a scripted transport - Support\Audio (PCM16 -> WAV); RecordingStore::discard() - provider binds RoomStore/RoomManager/RecordingStore/OpenAiRealtimeBridge and guards webrtc:serve behind class_exists(laravel-ws + reactphp-kernel) - laravel-ws + reactphp-kernel + react/* -> require-dev; runtime needs only illuminate + textalk/websocket - 40 tests green Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blax Software
parent
1be40ba59c
commit
9c791a8bba
58
README.md
58
README.md
|
|
@ -19,7 +19,8 @@ WebRTC for Laravel on the shared [`reactphp-kernel`](https://github.com/blax-sof
|
|||
- ⏺️ **Record the full call** — the browser streams its mic over the WebSocket and the server stores each participant's audio (`FileRecordingStore` → `<room>/<peer>.webm`); + a call-event log seam for auditing even unrecorded calls
|
||||
- 🧩 **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
|
||||
- 🤖 **AI realtime, provider hidden** — `OpenAiRealtimeBridge` relays a caller to OpenAI's Realtime API **server-side** (PCM in → spoken reply out, recordable), so the browser only ever talks to *you* and the model is a config swap
|
||||
- 🔀 **Use it on any transport** — the domain layer (rooms, recording, the turn-based realtime bridge) is transport-agnostic: run it on the bundled ReactPHP server *or* drive it from a WebSocket you already have; a swappable `RoomStore` makes membership Cache-backed for fork-per-message hosts
|
||||
- 🔌 **Pluggable engine** — `NullMediaEngine` (signaling-only) now; `Str0mMediaEngine` (Rust `str0m` via `ext-php-rs`, shipped in `rust/`) for real server-terminated media
|
||||
|
||||
## Installation
|
||||
|
|
@ -29,7 +30,7 @@ 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) + [`blax-software/laravel-ws`](https://github.com/blax-software/laravel-ws), resolved from git.blax.at via the `repositories` entry in `composer.json`.
|
||||
The domain layer (rooms, recording, the realtime bridge) needs only `illuminate/*` + `textalk/websocket`. The **bundled ReactPHP server** (`webrtc:serve`) additionally uses [`reactphp-kernel`](https://github.com/blax-software/reactphp-kernel) + [`laravel-ws`](https://github.com/blax-software/laravel-ws) — these are `require-dev`, so a host that runs the relay on its own WebSocket doesn't pull them (see [Use it on your own WebSocket](#use-it-on-your-own-websocket-no-bundled-server)).
|
||||
|
||||
## Quick Start
|
||||
|
||||
|
|
@ -133,6 +134,38 @@ WEBRTC_MEDIA_ENGINE="Blax\WebRtc\Media\Str0mMediaEngine"
|
|||
|
||||
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`.
|
||||
|
||||
### Use it on your own WebSocket (no bundled server)
|
||||
|
||||
The domain layer is transport-agnostic, so a host that already runs a WebSocket — including a classic **fork-per-message** Laravel WS — can drive the relay itself and skip this package's ReactPHP server entirely. Resolve the pieces and relay with your own transport:
|
||||
|
||||
```php
|
||||
use Blax\WebRtc\Rooms\RoomManager;
|
||||
use Blax\WebRtc\Contracts\RecordingStore;
|
||||
use Blax\WebRtc\Realtime\OpenAiRealtimeBridge;
|
||||
|
||||
// Typed rooms + membership — join() returns the peers already there to notify:
|
||||
$others = app(RoomManager::class)->join($room, $peerId);
|
||||
foreach ($others as $peer) $yourWs->sendTo($peer, ['type' => 'peer-joined', 'peer' => $peerId]);
|
||||
|
||||
// Store the call (your transport streams the mic to you however it likes):
|
||||
app(RecordingStore::class)->append($room, $peerId, $chunk);
|
||||
|
||||
// An AI joins the call — provider hidden, turn-based (one round-trip per push-to-talk):
|
||||
$turn = app(OpenAiRealtimeBridge::class)->exchange($pcm, [
|
||||
'api_key' => config('services.openai.key'), 'voice' => 'alloy', 'instructions' => '…',
|
||||
]);
|
||||
// $turn->pcm = PCM16 24kHz reply, $turn->transcript = text — play + record it.
|
||||
```
|
||||
|
||||
Membership lives in a swappable `RoomStore`. The default `ArrayRoomStore` keeps it in process memory (right for the long-lived bundled server); a fork-per-message host, which has no shared memory across frames, binds a Cache/Redis-backed one so state survives across workers:
|
||||
|
||||
```php
|
||||
// config/webrtc.php
|
||||
'room_store' => \App\Support\CacheRoomStore::class, // implements Blax\WebRtc\Contracts\RoomStore
|
||||
```
|
||||
|
||||
`OpenAiRealtimeBridge` talks to a `RealtimeTransport` (textalk/websocket by default), so it's unit-testable with a scripted fake. This is exactly how learn-atc runs the relay + a recorded OpenAI ATC participant on its existing WebSocket, without `laravel-ws`/`reactphp-kernel` installed.
|
||||
|
||||
## Configuration
|
||||
|
||||
`config/webrtc.php` covers the signaling bind (`host`/`port`/`tls`), `recording` (`enabled`/`path`/`format`), the `authorizer` (room join gate) and `events` (call log) bindings, the `media_engine`, advertised `ice_servers`, and the external `bridge` (e.g. OpenAI model + key).
|
||||
|
|
@ -144,14 +177,23 @@ The engine-backed path uses `Signaling\SignalingHandler` (offer/ice/record/conne
|
|||
## Architecture
|
||||
|
||||
```
|
||||
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/)
|
||||
laravel-webrtc (this)
|
||||
├─ Domain layer (transport-agnostic — needs no bundled server)
|
||||
│ ├─ RoomManager + RoomStore typed rooms, membership, presence
|
||||
│ ├─ RecordingStore per-participant call recording
|
||||
│ └─ OpenAiRealtimeBridge AI joins the call, provider hidden
|
||||
│
|
||||
├─ Bundled server (require-dev) run it for you…
|
||||
│ reactphp-kernel shared backbone (loop, IPC, signals)
|
||||
│ └─ laravel-ws WebSocket transport (RFC6455)
|
||||
│ └─ RelaySignalingHandler browser P2P calls over the loop
|
||||
│
|
||||
└─ MediaEngine (optional) …or terminate media server-side
|
||||
└─ Str0mMediaEngine Rust/str0m via ext-php-rs (rust/)
|
||||
```
|
||||
|
||||
The domain layer runs on **any** transport (the bundled server, or a host WS like learn-atc's fork-per-message WebSocket). The bundled server + media engine are opt-in.
|
||||
|
||||
## Testing
|
||||
|
||||
```bash
|
||||
|
|
|
|||
|
|
@ -12,16 +12,17 @@
|
|||
],
|
||||
"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",
|
||||
"illuminate/console": "^10.0|^11.0|^12.0",
|
||||
"illuminate/support": "^10.0|^11.0|^12.0",
|
||||
"illuminate/console": "^10.0|^11.0|^12.0"
|
||||
"textalk/websocket": "^1.5|^1.6"
|
||||
},
|
||||
"require-dev": {
|
||||
"blax-software/laravel-ws": "dev-master",
|
||||
"blax-software/reactphp-kernel": "dev-master",
|
||||
"laravel/pint": "^1.13",
|
||||
"phpunit/phpunit": "^10.5|^11.0",
|
||||
"laravel/pint": "^1.13"
|
||||
"react/event-loop": "^1.5",
|
||||
"react/socket": "^1.15"
|
||||
},
|
||||
"suggest": {
|
||||
"ext-ffi": "Required by the Str0m media engine (Rust core exposed to PHP)."
|
||||
|
|
|
|||
|
|
@ -57,6 +57,18 @@ return [
|
|||
'authorizer' => env('WEBRTC_AUTHORIZER', DefaultRoomAuthorizer::class),
|
||||
'events' => env('WEBRTC_EVENTS', NullCallEventListener::class),
|
||||
|
||||
/*
|
||||
|--------------------------------------------------------------------------
|
||||
| Room membership store
|
||||
|--------------------------------------------------------------------------
|
||||
| Where room membership + presence lives. The default ArrayRoomStore keeps it
|
||||
| in process memory (right for the long-lived bundled server). A fork-per-
|
||||
| message host (a classic Laravel WebSocket handling each frame in a fresh
|
||||
| worker) binds a Cache/Redis-backed RoomStore so state survives across
|
||||
| workers — that is how a host runs this relay on its existing WS.
|
||||
*/
|
||||
'room_store' => env('WEBRTC_ROOM_STORE', \Blax\WebRtc\Rooms\Stores\ArrayRoomStore::class),
|
||||
|
||||
/*
|
||||
|--------------------------------------------------------------------------
|
||||
| ICE servers (STUN/TURN) advertised to browsers
|
||||
|
|
@ -71,13 +83,18 @@ return [
|
|||
|--------------------------------------------------------------------------
|
||||
| External realtime bridge (provider hidden from the browser)
|
||||
|--------------------------------------------------------------------------
|
||||
| The media engine dials these; the browser only ever talks to us, so the AI
|
||||
| provider is invisible and the model is swappable server-side.
|
||||
| OpenAiRealtimeBridge (or the media engine) dials these; the browser only ever
|
||||
| talks to us, so the AI provider is invisible and the model is swappable
|
||||
| server-side. `instructions` is the persona (the host injects its own — e.g.
|
||||
| an ATC prompt) and `voice` the spoken voice.
|
||||
*/
|
||||
'bridge' => [
|
||||
'openai' => [
|
||||
'endpoint' => env('WEBRTC_OPENAI_ENDPOINT', 'wss://api.openai.com/v1/realtime'),
|
||||
'model' => env('WEBRTC_OPENAI_MODEL', 'gpt-realtime'),
|
||||
'api_key' => env('OPENAI_API_KEY'),
|
||||
'voice' => env('WEBRTC_OPENAI_VOICE', 'alloy'),
|
||||
'instructions' => null,
|
||||
],
|
||||
],
|
||||
];
|
||||
|
|
|
|||
|
|
@ -0,0 +1,24 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Contracts;
|
||||
|
||||
use Blax\WebRtc\Realtime\OpenAiRealtimeBridge;
|
||||
|
||||
/**
|
||||
* A bidirectional text-frame link to an external realtime AI provider (its
|
||||
* WebSocket transport). Abstracted so {@see OpenAiRealtimeBridge}
|
||||
* is unit-testable with a scripted fake and swappable between blocking
|
||||
* (textalk/websocket) and loop-based clients.
|
||||
*/
|
||||
interface RealtimeTransport
|
||||
{
|
||||
/** Send one JSON event to the provider. */
|
||||
public function send(string $text): void;
|
||||
|
||||
/** Receive the next event, or null on timeout / close. */
|
||||
public function receive(): ?string;
|
||||
|
||||
public function close(): void;
|
||||
}
|
||||
|
|
@ -20,4 +20,7 @@ interface RecordingStore
|
|||
* @return string|null a locator (path/URL) for the finished recording, or null
|
||||
*/
|
||||
public function finalize(string $roomId, string $peerId): ?string;
|
||||
|
||||
/** Discard a participant's recording so a new one starts clean (e.g. re-join / restart). */
|
||||
public function discard(string $roomId, string $peerId): void;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,58 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Contracts;
|
||||
|
||||
use Blax\WebRtc\Rooms\RoomManager;
|
||||
use Blax\WebRtc\Rooms\Stores\ArrayRoomStore;
|
||||
|
||||
/**
|
||||
* Where room membership + presence lives. The default {@see ArrayRoomStore}
|
||||
* keeps it in process memory, which is right for the long-lived bundled server
|
||||
* (one process owns every connection). A fork-per-message host (e.g. a classic
|
||||
* Laravel WebSocket that handles each frame in a fresh worker) has no shared
|
||||
* memory across messages, so it binds a store backed by Cache/Redis instead —
|
||||
* that is how learn-atc runs the relay on its existing WS without this package's
|
||||
* ReactPHP server.
|
||||
*
|
||||
* Peers keep exactly one room (joining a new one leaves the old). `add`/`remove`
|
||||
* must also maintain the peer→room reverse index that {@see roomOf} reads.
|
||||
*/
|
||||
interface RoomStore
|
||||
{
|
||||
/**
|
||||
* Ordered peer ids currently in the room (insertion order).
|
||||
*
|
||||
* @return array<int,string>
|
||||
*/
|
||||
public function members(string $roomId): array;
|
||||
|
||||
/**
|
||||
* Presence info a peer published on join (empty array if none / not a presence room).
|
||||
*
|
||||
* @return array<string,mixed>
|
||||
*/
|
||||
public function info(string $roomId, string $peerId): array;
|
||||
|
||||
/** The room a peer is currently in, or null. */
|
||||
public function roomOf(string $peerId): ?string;
|
||||
|
||||
/**
|
||||
* Add/update a peer (with presence info) in a room and set its reverse index.
|
||||
*
|
||||
* @param array<string,mixed> $info
|
||||
*/
|
||||
public function add(string $roomId, string $peerId, array $info): void;
|
||||
|
||||
/** Remove a peer from a room, clear its reverse index, and drop the room if it empties. */
|
||||
public function remove(string $roomId, string $peerId): void;
|
||||
|
||||
/**
|
||||
* All non-empty room ids. Best-effort: a store that can't enumerate (e.g. a bare
|
||||
* Cache backend) may return an empty array — only {@see RoomManager::count()} relies on it.
|
||||
*
|
||||
* @return array<int,string>
|
||||
*/
|
||||
public function roomIds(): array;
|
||||
}
|
||||
|
|
@ -0,0 +1,136 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Realtime;
|
||||
|
||||
use Blax\WebRtc\Contracts\RealtimeTransport;
|
||||
|
||||
/**
|
||||
* Bridges a call to OpenAI's GA Realtime API over its WebSocket transport, so the
|
||||
* SERVER — not the browser — talks to the provider. The provider is invisible to
|
||||
* the client, the model is swappable server-side, and the audio can be recorded.
|
||||
*
|
||||
* Turn-based: {@see exchange()} takes the caller's PCM utterance, drives one
|
||||
* OpenAI session (session.update → input_audio_buffer → response), and returns the
|
||||
* spoken reply as PCM. That fits both the long-lived server and a fork-per-message
|
||||
* host, where each turn runs inside a single request.
|
||||
*
|
||||
* Audio is PCM16 mono 24kHz in both directions (OpenAI's realtime wire format);
|
||||
* transcode mic audio (webm/opus, …) to that before calling and back for playback.
|
||||
*
|
||||
* GA WebSocket gotchas baked in: NO "OpenAI-Beta: realtime=v1" header (routes to
|
||||
* the retired beta), output_modalities must be ['audio'] OR ['text'] (not both),
|
||||
* audio arrives as response.output_audio.delta (base64 PCM16).
|
||||
*/
|
||||
final class OpenAiRealtimeBridge
|
||||
{
|
||||
/** @var \Closure(string,array<string,string>,int):RealtimeTransport */
|
||||
private \Closure $transportFactory;
|
||||
|
||||
public function __construct(
|
||||
?callable $transportFactory = null,
|
||||
private string $endpoint = 'wss://api.openai.com/v1/realtime',
|
||||
) {
|
||||
$this->transportFactory = $transportFactory !== null
|
||||
? \Closure::fromCallable($transportFactory)
|
||||
: static fn (string $url, array $headers, int $timeout): RealtimeTransport => new TextalkRealtimeTransport($url, $headers, $timeout);
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one turn: send $inputPcm to OpenAI realtime and return the spoken reply.
|
||||
*
|
||||
* @param string $inputPcm Raw PCM16 mono 24kHz little-endian.
|
||||
* @param array{api_key?:string,model?:string,voice?:string,instructions?:string,timeout?:int} $opts
|
||||
*/
|
||||
public function exchange(string $inputPcm, array $opts): RealtimeTurn
|
||||
{
|
||||
$key = (string) ($opts['api_key'] ?? '');
|
||||
if ($key === '') {
|
||||
return new RealtimeTurn(error: 'missing api_key');
|
||||
}
|
||||
if ($inputPcm === '') {
|
||||
return new RealtimeTurn(error: 'empty input audio');
|
||||
}
|
||||
|
||||
$model = (string) ($opts['model'] ?? 'gpt-realtime');
|
||||
$voice = (string) ($opts['voice'] ?? 'alloy');
|
||||
$instructions = (string) ($opts['instructions'] ?? '');
|
||||
$timeout = (int) ($opts['timeout'] ?? 30);
|
||||
|
||||
try {
|
||||
$transport = ($this->transportFactory)(
|
||||
"{$this->endpoint}?model={$model}",
|
||||
['Authorization' => 'Bearer '.$key],
|
||||
$timeout,
|
||||
);
|
||||
} catch (\Throwable $e) {
|
||||
return new RealtimeTurn(error: 'connect failed: '.$e->getMessage());
|
||||
}
|
||||
|
||||
$pcm = '';
|
||||
$transcript = '';
|
||||
$error = null;
|
||||
$seen = [];
|
||||
|
||||
try {
|
||||
// Manual push-to-talk: turn_detection null, we commit the buffer. Audio
|
||||
// in/out default to PCM16 24kHz mono, so no explicit format is needed.
|
||||
$session = [
|
||||
'type' => 'realtime',
|
||||
'audio' => [
|
||||
'input' => ['turn_detection' => null],
|
||||
'output' => ['voice' => $voice],
|
||||
],
|
||||
'output_modalities' => ['audio'],
|
||||
];
|
||||
if ($instructions !== '') {
|
||||
$session['instructions'] = $instructions;
|
||||
}
|
||||
$transport->send((string) json_encode(['type' => 'session.update', 'session' => $session]));
|
||||
|
||||
// Stream the input audio in. base64 is split on 4-char boundaries
|
||||
// (65536 is a multiple of 4) so each frame stays valid.
|
||||
foreach (str_split(base64_encode($inputPcm), 65536) as $chunk) {
|
||||
$transport->send((string) json_encode(['type' => 'input_audio_buffer.append', 'audio' => $chunk]));
|
||||
}
|
||||
$transport->send((string) json_encode(['type' => 'input_audio_buffer.commit']));
|
||||
$transport->send((string) json_encode(['type' => 'response.create']));
|
||||
|
||||
$deadline = time() + $timeout;
|
||||
while (time() < $deadline) {
|
||||
$raw = $transport->receive();
|
||||
if ($raw === null) {
|
||||
break; // timeout / closed
|
||||
}
|
||||
if ($raw === '') {
|
||||
continue;
|
||||
}
|
||||
|
||||
$evt = json_decode($raw, true);
|
||||
if (! is_array($evt)) {
|
||||
continue;
|
||||
}
|
||||
$type = (string) ($evt['type'] ?? '?');
|
||||
$seen[$type] = ($seen[$type] ?? 0) + 1;
|
||||
|
||||
if (in_array($type, ['response.audio.delta', 'response.output_audio.delta'], true)) {
|
||||
$pcm .= base64_decode((string) ($evt['delta'] ?? ''));
|
||||
} elseif (in_array($type, ['response.audio_transcript.delta', 'response.output_audio_transcript.delta'], true)) {
|
||||
$transcript .= (string) ($evt['delta'] ?? '');
|
||||
} elseif ($type === 'error') {
|
||||
$error = $evt['error']['message'] ?? json_encode($evt['error'] ?? $evt);
|
||||
break;
|
||||
} elseif ($type === 'response.done') {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (\Throwable $e) {
|
||||
$error = $e->getMessage();
|
||||
} finally {
|
||||
$transport->close();
|
||||
}
|
||||
|
||||
return new RealtimeTurn(pcm: $pcm, transcript: trim($transcript), error: $error, events: $seen);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Realtime;
|
||||
|
||||
/**
|
||||
* The result of one realtime AI turn: the spoken reply as raw PCM16 (mono 24kHz,
|
||||
* the provider's wire format), its transcript, an error if the turn failed, and a
|
||||
* per-event-type tally for diagnostics.
|
||||
*/
|
||||
final class RealtimeTurn
|
||||
{
|
||||
/** @param array<string,int> $events event type => count */
|
||||
public function __construct(
|
||||
public readonly string $pcm = '',
|
||||
public readonly string $transcript = '',
|
||||
public readonly ?string $error = null,
|
||||
public readonly array $events = [],
|
||||
) {}
|
||||
|
||||
public function ok(): bool
|
||||
{
|
||||
return $this->error === null && $this->pcm !== '';
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Realtime;
|
||||
|
||||
use Blax\WebRtc\Contracts\RealtimeTransport;
|
||||
use WebSocket\Client as WebSocketClient;
|
||||
|
||||
/**
|
||||
* {@see RealtimeTransport} over textalk/websocket (a blocking WS client). This is
|
||||
* the default transport for {@see OpenAiRealtimeBridge}: fine for a turn-based
|
||||
* bridge where each turn opens, drives and closes one provider session, and it
|
||||
* fits a fork-per-message host (the whole turn runs inside one request/worker).
|
||||
*/
|
||||
final class TextalkRealtimeTransport implements RealtimeTransport
|
||||
{
|
||||
private WebSocketClient $client;
|
||||
|
||||
/** @param array<string,string> $headers */
|
||||
public function __construct(string $url, array $headers, int $timeout)
|
||||
{
|
||||
$this->client = new WebSocketClient($url, [
|
||||
'headers' => $headers,
|
||||
'timeout' => $timeout,
|
||||
]);
|
||||
}
|
||||
|
||||
public function send(string $text): void
|
||||
{
|
||||
$this->client->text($text);
|
||||
}
|
||||
|
||||
public function receive(): ?string
|
||||
{
|
||||
try {
|
||||
$raw = $this->client->receive();
|
||||
} catch (\Throwable) {
|
||||
return null; // timeout / closed
|
||||
}
|
||||
|
||||
return is_string($raw) ? $raw : null;
|
||||
}
|
||||
|
||||
public function close(): void
|
||||
{
|
||||
try {
|
||||
$this->client->close();
|
||||
} catch (\Throwable) {
|
||||
// ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -54,6 +54,20 @@ final class FileRecordingStore implements RecordingStore
|
|||
return $this->path($roomId, $peerId);
|
||||
}
|
||||
|
||||
public function discard(string $roomId, string $peerId): void
|
||||
{
|
||||
$key = $this->key($roomId, $peerId);
|
||||
if (isset($this->handles[$key])) {
|
||||
fclose($this->handles[$key]);
|
||||
unset($this->handles[$key]);
|
||||
}
|
||||
|
||||
$path = $this->path($roomId, $peerId);
|
||||
if (is_file($path)) {
|
||||
@unlink($path);
|
||||
}
|
||||
}
|
||||
|
||||
/** The path a recording is (or will be) written to. */
|
||||
public function path(string $roomId, string $peerId): string
|
||||
{
|
||||
|
|
|
|||
|
|
@ -4,23 +4,37 @@ declare(strict_types=1);
|
|||
|
||||
namespace Blax\WebRtc\Rooms;
|
||||
|
||||
use Blax\WebRtc\Contracts\RoomStore;
|
||||
use Blax\WebRtc\Rooms\Stores\ArrayRoomStore;
|
||||
|
||||
/**
|
||||
* Tracks which peers are in which room for party-to-party / group calls, and
|
||||
* answers the mesh question the SignalingHandler needs: "who is a new joiner
|
||||
* answers the mesh question the signaling layer needs: "who is a new joiner
|
||||
* already in a call with?" Empty rooms are dropped automatically.
|
||||
*
|
||||
* Membership lives in a pluggable {@see RoomStore}: the default {@see ArrayRoomStore}
|
||||
* keeps it in memory (bundled server), while a fork-per-message host binds a
|
||||
* Cache/Redis-backed store so state survives across workers. The public API is
|
||||
* unchanged either way.
|
||||
*/
|
||||
final class RoomManager
|
||||
{
|
||||
/** @var array<string,Room> roomId => Room */
|
||||
private array $rooms = [];
|
||||
private RoomStore $store;
|
||||
|
||||
/** @var array<string,string> peerId => roomId */
|
||||
private array $peerRoom = [];
|
||||
public function __construct(?RoomStore $store = null)
|
||||
{
|
||||
$this->store = $store ?? new ArrayRoomStore;
|
||||
}
|
||||
|
||||
public function store(): RoomStore
|
||||
{
|
||||
return $this->store;
|
||||
}
|
||||
|
||||
/**
|
||||
* Join $peerId to $roomId (with optional presence $info), returning the OTHER
|
||||
* peers already present so the caller can mesh / notify them. The room's type
|
||||
* is derived from its name (RoomType::fromName) on first join.
|
||||
* is derived from its name (RoomType::fromName) by {@see room()}.
|
||||
*
|
||||
* @param array<string,mixed> $info
|
||||
* @return array<int,string>
|
||||
|
|
@ -28,13 +42,13 @@ final class RoomManager
|
|||
public function join(string $roomId, string $peerId, array $info = []): array
|
||||
{
|
||||
// Moving between rooms: leave the old one first.
|
||||
if (($this->peerRoom[$peerId] ?? null) !== null && $this->peerRoom[$peerId] !== $roomId) {
|
||||
$current = $this->store->roomOf($peerId);
|
||||
if ($current !== null && $current !== $roomId) {
|
||||
$this->leave($peerId);
|
||||
}
|
||||
|
||||
$room = $this->rooms[$roomId] ??= new Room($roomId, RoomType::fromName($roomId));
|
||||
$others = $room->add($peerId, $info);
|
||||
$this->peerRoom[$peerId] = $roomId;
|
||||
$others = $this->store->members($roomId);
|
||||
$this->store->add($roomId, $peerId, $info);
|
||||
|
||||
return $others;
|
||||
}
|
||||
|
|
@ -47,42 +61,44 @@ final class RoomManager
|
|||
*/
|
||||
public function leave(string $peerId): array
|
||||
{
|
||||
$roomId = $this->peerRoom[$peerId] ?? null;
|
||||
if ($roomId === null || ! isset($this->rooms[$roomId])) {
|
||||
$roomId = $this->store->roomOf($peerId);
|
||||
if ($roomId === null) {
|
||||
return [];
|
||||
}
|
||||
|
||||
$room = $this->rooms[$roomId];
|
||||
$room->remove($peerId);
|
||||
unset($this->peerRoom[$peerId]);
|
||||
$this->store->remove($roomId, $peerId);
|
||||
|
||||
if ($room->isEmpty()) {
|
||||
unset($this->rooms[$roomId]);
|
||||
|
||||
return [];
|
||||
}
|
||||
|
||||
return $room->peers();
|
||||
return $this->store->members($roomId);
|
||||
}
|
||||
|
||||
public function roomOf(string $peerId): ?string
|
||||
{
|
||||
return $this->peerRoom[$peerId] ?? null;
|
||||
return $this->store->roomOf($peerId);
|
||||
}
|
||||
|
||||
public function room(string $roomId): ?Room
|
||||
{
|
||||
return $this->rooms[$roomId] ?? null;
|
||||
$members = $this->store->members($roomId);
|
||||
if ($members === []) {
|
||||
return null;
|
||||
}
|
||||
|
||||
$room = new Room($roomId, RoomType::fromName($roomId));
|
||||
foreach ($members as $peerId) {
|
||||
$room->add($peerId, $this->store->info($roomId, $peerId));
|
||||
}
|
||||
|
||||
return $room;
|
||||
}
|
||||
|
||||
/** @return array<int,string> */
|
||||
public function peersIn(string $roomId): array
|
||||
{
|
||||
return isset($this->rooms[$roomId]) ? $this->rooms[$roomId]->peers() : [];
|
||||
return $this->store->members($roomId);
|
||||
}
|
||||
|
||||
public function count(): int
|
||||
{
|
||||
return count($this->rooms);
|
||||
return count($this->store->roomIds());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,56 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Rooms\Stores;
|
||||
|
||||
use Blax\WebRtc\Contracts\RoomStore;
|
||||
|
||||
/**
|
||||
* In-process room store: membership + presence in PHP arrays. Right for the
|
||||
* long-lived bundled server where one process owns every connection. A
|
||||
* fork-per-message host binds a Cache-backed store instead (see {@see RoomStore}).
|
||||
*/
|
||||
final class ArrayRoomStore implements RoomStore
|
||||
{
|
||||
/** @var array<string,array<string,array<string,mixed>>> roomId => (peerId => presence info), insertion-ordered */
|
||||
private array $rooms = [];
|
||||
|
||||
/** @var array<string,string> peerId => roomId */
|
||||
private array $peerRoom = [];
|
||||
|
||||
public function members(string $roomId): array
|
||||
{
|
||||
return array_keys($this->rooms[$roomId] ?? []);
|
||||
}
|
||||
|
||||
public function info(string $roomId, string $peerId): array
|
||||
{
|
||||
return $this->rooms[$roomId][$peerId] ?? [];
|
||||
}
|
||||
|
||||
public function roomOf(string $peerId): ?string
|
||||
{
|
||||
return $this->peerRoom[$peerId] ?? null;
|
||||
}
|
||||
|
||||
public function add(string $roomId, string $peerId, array $info): void
|
||||
{
|
||||
$this->rooms[$roomId][$peerId] = $info;
|
||||
$this->peerRoom[$peerId] = $roomId;
|
||||
}
|
||||
|
||||
public function remove(string $roomId, string $peerId): void
|
||||
{
|
||||
unset($this->rooms[$roomId][$peerId], $this->peerRoom[$peerId]);
|
||||
|
||||
if (($this->rooms[$roomId] ?? []) === []) {
|
||||
unset($this->rooms[$roomId]);
|
||||
}
|
||||
}
|
||||
|
||||
public function roomIds(): array
|
||||
{
|
||||
return array_keys($this->rooms);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Support;
|
||||
|
||||
/**
|
||||
* Small audio helpers for the realtime bridge. Providers speak raw PCM16; browsers
|
||||
* play containers — {@see wav()} wraps PCM as a WAV so it plays in an <audio> tag or
|
||||
* a data: URL with no client-side decoding.
|
||||
*/
|
||||
final class Audio
|
||||
{
|
||||
/** Wrap raw PCM16 little-endian as a WAV container (default 24kHz mono, the realtime format). */
|
||||
public static function wav(string $pcm, int $sampleRate = 24000, int $channels = 1, int $bits = 16): string
|
||||
{
|
||||
$byteRate = $sampleRate * $channels * intdiv($bits, 8);
|
||||
$blockAlign = $channels * intdiv($bits, 8);
|
||||
|
||||
return 'RIFF'.pack('V', 36 + strlen($pcm)).'WAVE'
|
||||
.'fmt '.pack('V', 16).pack('v', 1).pack('v', $channels)
|
||||
.pack('V', $sampleRate).pack('V', $byteRate).pack('v', $blockAlign).pack('v', $bits)
|
||||
.'data'.pack('V', strlen($pcm)).$pcm;
|
||||
}
|
||||
|
||||
/** A `data:audio/wav;base64,…` URL for raw PCM16, ready to hand a browser <audio> element. */
|
||||
public static function wavDataUri(string $pcm, int $sampleRate = 24000, int $channels = 1, int $bits = 16): string
|
||||
{
|
||||
return 'data:audio/wav;base64,'.base64_encode(self::wav($pcm, $sampleRate, $channels, $bits));
|
||||
}
|
||||
}
|
||||
|
|
@ -4,13 +4,21 @@ declare(strict_types=1);
|
|||
|
||||
namespace Blax\WebRtc;
|
||||
|
||||
use Blax\ReactPhpKernel\Kernel;
|
||||
use Blax\WebRtc\Authorization\DefaultRoomAuthorizer;
|
||||
use Blax\WebRtc\Console\Commands\StartWebRtcServer;
|
||||
use Blax\WebRtc\Contracts\CallEventListener;
|
||||
use Blax\WebRtc\Contracts\MediaEngine;
|
||||
use Blax\WebRtc\Contracts\RecordingStore;
|
||||
use Blax\WebRtc\Contracts\RoomAuthorizer;
|
||||
use Blax\WebRtc\Contracts\RoomStore;
|
||||
use Blax\WebRtc\Events\NullCallEventListener;
|
||||
use Blax\WebRtc\Media\NullMediaEngine;
|
||||
use Blax\WebRtc\Realtime\OpenAiRealtimeBridge;
|
||||
use Blax\WebRtc\Recording\FileRecordingStore;
|
||||
use Blax\WebRtc\Rooms\RoomManager;
|
||||
use Blax\WebRtc\Rooms\Stores\ArrayRoomStore;
|
||||
use Blax\Ws\WebSocketServer;
|
||||
use Illuminate\Support\ServiceProvider;
|
||||
|
||||
class WebRtcServiceProvider extends ServiceProvider
|
||||
|
|
@ -33,6 +41,33 @@ class WebRtcServiceProvider extends ServiceProvider
|
|||
$this->app->bind(CallEventListener::class, function ($app) {
|
||||
return $app->make((string) config('webrtc.events', NullCallEventListener::class));
|
||||
});
|
||||
|
||||
// Room membership store — swappable so a fork-per-message host (learn-atc's
|
||||
// existing WS) can back it with Cache/Redis instead of process memory. The
|
||||
// RoomManager is per-resolve so a fresh worker reads current state.
|
||||
$this->app->bind(RoomStore::class, function ($app) {
|
||||
return $app->make((string) config('webrtc.room_store', ArrayRoomStore::class));
|
||||
});
|
||||
$this->app->bind(RoomManager::class, function ($app) {
|
||||
return new RoomManager($app->make(RoomStore::class));
|
||||
});
|
||||
|
||||
// Where a call's audio is stored (per participant). Host may override the class.
|
||||
$this->app->bind(RecordingStore::class, function () {
|
||||
$recording = (array) config('webrtc.recording', []);
|
||||
|
||||
return new FileRecordingStore(
|
||||
(string) ($recording['path'] ?? 'webrtc'),
|
||||
(string) ($recording['format'] ?? 'webm'),
|
||||
);
|
||||
});
|
||||
|
||||
// Realtime AI bridge (provider hidden from the browser, model swappable server-side).
|
||||
$this->app->bind(OpenAiRealtimeBridge::class, function () {
|
||||
$endpoint = (string) config('webrtc.bridge.openai.endpoint', 'wss://api.openai.com/v1/realtime');
|
||||
|
||||
return new OpenAiRealtimeBridge(endpoint: $endpoint);
|
||||
});
|
||||
}
|
||||
|
||||
public function boot(): void
|
||||
|
|
@ -42,9 +77,16 @@ class WebRtcServiceProvider extends ServiceProvider
|
|||
__DIR__.'/../config/webrtc.php' => $this->app->configPath('webrtc.php'),
|
||||
], 'webrtc-config');
|
||||
|
||||
$this->commands([
|
||||
StartWebRtcServer::class,
|
||||
]);
|
||||
// The bundled ReactPHP relay server needs blax-software/laravel-ws +
|
||||
// reactphp-kernel (dev-only deps). A host that only consumes the domain
|
||||
// layer (rooms/recording/realtime bridge) on its OWN transport won't have
|
||||
// them — checking the dependency classes first avoids autoloading the
|
||||
// command (and its imports) when they're absent.
|
||||
if (class_exists(WebSocketServer::class) && class_exists(Kernel::class)) {
|
||||
$this->commands([
|
||||
StartWebRtcServer::class,
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,4 +30,9 @@ final class MemoryRecordingStore implements RecordingStore
|
|||
|
||||
return "mem://{$key}";
|
||||
}
|
||||
|
||||
public function discard(string $roomId, string $peerId): void
|
||||
{
|
||||
unset($this->data["{$roomId}|{$peerId}"]);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,40 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Tests\Fixtures;
|
||||
|
||||
use Blax\WebRtc\Contracts\RealtimeTransport;
|
||||
|
||||
/** A RealtimeTransport double: records sent frames and replays a scripted inbound sequence. */
|
||||
final class ScriptedRealtimeTransport implements RealtimeTransport
|
||||
{
|
||||
/** @var array<int,string> */
|
||||
public array $sent = [];
|
||||
|
||||
public bool $closed = false;
|
||||
|
||||
/** @param array<int,string> $inbound scripted JSON frames handed back by receive() */
|
||||
public function __construct(private array $inbound) {}
|
||||
|
||||
public function send(string $text): void
|
||||
{
|
||||
$this->sent[] = $text;
|
||||
}
|
||||
|
||||
public function receive(): ?string
|
||||
{
|
||||
return array_shift($this->inbound); // null once exhausted
|
||||
}
|
||||
|
||||
public function close(): void
|
||||
{
|
||||
$this->closed = true;
|
||||
}
|
||||
|
||||
/** @return array<int,string> the `type` of each sent frame, in order */
|
||||
public function sentTypes(): array
|
||||
{
|
||||
return array_map(static fn ($s) => (string) (json_decode($s, true)['type'] ?? '?'), $this->sent);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Tests;
|
||||
|
||||
use Blax\WebRtc\Realtime\OpenAiRealtimeBridge;
|
||||
use Blax\WebRtc\Tests\Fixtures\ScriptedRealtimeTransport;
|
||||
use PHPUnit\Framework\TestCase;
|
||||
|
||||
final class OpenAiRealtimeBridgeTest extends TestCase
|
||||
{
|
||||
public function test_exchange_drives_the_session_and_assembles_the_reply(): void
|
||||
{
|
||||
$inbound = [
|
||||
(string) json_encode(['type' => 'session.updated']),
|
||||
(string) json_encode(['type' => 'response.output_audio_transcript.delta', 'delta' => 'Roger, ']),
|
||||
(string) json_encode(['type' => 'response.output_audio.delta', 'delta' => base64_encode('AB')]),
|
||||
(string) json_encode(['type' => 'response.output_audio_transcript.delta', 'delta' => 'wilco.']),
|
||||
(string) json_encode(['type' => 'response.output_audio.delta', 'delta' => base64_encode('CD')]),
|
||||
(string) json_encode(['type' => 'response.done']),
|
||||
];
|
||||
$fake = new ScriptedRealtimeTransport($inbound);
|
||||
|
||||
$bridge = new OpenAiRealtimeBridge(fn ($url, $headers, $timeout) => $fake);
|
||||
$turn = $bridge->exchange('PILOTPCM', ['api_key' => 'sk-test', 'voice' => 'echo', 'instructions' => 'be ATC']);
|
||||
|
||||
$this->assertNull($turn->error);
|
||||
$this->assertTrue($turn->ok());
|
||||
$this->assertSame('ABCD', $turn->pcm);
|
||||
$this->assertSame('Roger, wilco.', $turn->transcript);
|
||||
$this->assertTrue($fake->closed);
|
||||
|
||||
// Correct GA protocol on the wire, in order.
|
||||
$types = $fake->sentTypes();
|
||||
$this->assertSame('session.update', $types[0]);
|
||||
$this->assertContains('input_audio_buffer.append', $types);
|
||||
$this->assertContains('input_audio_buffer.commit', $types);
|
||||
$this->assertContains('response.create', $types);
|
||||
|
||||
// Session config carries the voice + instructions and audio-only modality.
|
||||
$session = json_decode($fake->sent[0], true)['session'];
|
||||
$this->assertSame(['audio'], $session['output_modalities']);
|
||||
$this->assertSame('echo', $session['audio']['output']['voice']);
|
||||
$this->assertSame('be ATC', $session['instructions']);
|
||||
}
|
||||
|
||||
public function test_it_validates_before_dialing(): void
|
||||
{
|
||||
$bridge = new OpenAiRealtimeBridge(function () {
|
||||
throw new \RuntimeException('should not dial');
|
||||
});
|
||||
|
||||
$this->assertSame('missing api_key', $bridge->exchange('pcm', [])->error);
|
||||
$this->assertSame('empty input audio', $bridge->exchange('', ['api_key' => 'k'])->error);
|
||||
}
|
||||
|
||||
public function test_it_surfaces_provider_errors(): void
|
||||
{
|
||||
$inbound = [(string) json_encode(['type' => 'error', 'error' => ['message' => 'boom']])];
|
||||
$bridge = new OpenAiRealtimeBridge(fn ($url, $headers, $timeout) => new ScriptedRealtimeTransport($inbound));
|
||||
|
||||
$turn = $bridge->exchange('PCM', ['api_key' => 'k']);
|
||||
|
||||
$this->assertSame('boom', $turn->error);
|
||||
$this->assertFalse($turn->ok());
|
||||
$this->assertSame('', $turn->pcm);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace Blax\WebRtc\Tests;
|
||||
|
||||
use Blax\WebRtc\Rooms\RoomManager;
|
||||
use Blax\WebRtc\Rooms\Stores\ArrayRoomStore;
|
||||
use PHPUnit\Framework\TestCase;
|
||||
|
||||
final class RoomStoreTest extends TestCase
|
||||
{
|
||||
public function test_array_store_tracks_membership_presence_and_reverse_index(): void
|
||||
{
|
||||
$store = new ArrayRoomStore;
|
||||
$store->add('presence-lobby', 'p1', ['name' => 'Ada']);
|
||||
$store->add('presence-lobby', 'p2', []);
|
||||
|
||||
$this->assertSame(['p1', 'p2'], $store->members('presence-lobby'));
|
||||
$this->assertSame(['name' => 'Ada'], $store->info('presence-lobby', 'p1'));
|
||||
$this->assertSame('presence-lobby', $store->roomOf('p1'));
|
||||
$this->assertSame(['presence-lobby'], $store->roomIds());
|
||||
|
||||
$store->remove('presence-lobby', 'p1');
|
||||
$this->assertSame(['p2'], $store->members('presence-lobby'));
|
||||
$this->assertNull($store->roomOf('p1'));
|
||||
|
||||
$store->remove('presence-lobby', 'p2');
|
||||
$this->assertSame([], $store->roomIds()); // room dropped when empty
|
||||
}
|
||||
|
||||
public function test_room_manager_uses_the_injected_store_and_exposes_presence(): void
|
||||
{
|
||||
$store = new ArrayRoomStore;
|
||||
$rooms = new RoomManager($store);
|
||||
|
||||
$rooms->join('presence-x', 'p1', ['role' => 'host']);
|
||||
$this->assertSame(['p1'], $rooms->join('presence-x', 'p2', ['role' => 'guest']));
|
||||
|
||||
$room = $rooms->room('presence-x');
|
||||
$this->assertNotNull($room);
|
||||
$this->assertSame(['p1', 'p2'], $room->peers());
|
||||
$this->assertSame(['role' => 'host'], $room->info('p1'));
|
||||
$this->assertSame($store, $rooms->store());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue