laravel-webrtc/src/Media/Rust/ControlClient.php

<?php

declare(strict_types=1);

namespace Blax\WebRtc\Media\Rust;

/**
 * JSON-lines client for the Rust SFU sidecar's Unix control socket.
 *
 * One JSON object per line. Requests carry an `id` the reply echoes; the
 * sidecar also PUSHES notices (`{"event":"peer_left",...}`) onto every control
 * connection — those are buffered while waiting for a reply and retrievable
 * via drainEvents().
 */
class ControlClient
{
    /** @var resource|null */
    private $stream = null;

    private int $nextId = 1;

    /** @var list<array<string,mixed>> */
    private array $events = [];

    public function __construct(
        private readonly string $socketPath,
        private readonly float $timeout = 5.0,
    ) {}

    /**
     * Build a client on an already-open stream (tests use a socket pair).
     *
     * @param  resource  $stream
     */
    public static function fromStream($stream): self
    {
        $client = new self('');
        $client->stream = $stream;

        return $client;
    }

    /**
     * Send one command and wait for its reply. Throws on transport errors and
     * on `ok:false` replies.
     *
     * @param  array<string,mixed>  $params
     * @return array<string,mixed> the decoded reply (includes `ok` => true)
     */
    public function request(string $cmd, array $params = []): array
    {
        $id = $this->nextId++;
        $stream = $this->stream();

        $json = json_encode(['id' => $id, 'cmd' => $cmd] + $params, JSON_UNESCAPED_SLASHES);
        if (@fwrite($stream, $json."\n") === false) {
            $this->close();
            throw new SidecarException("sidecar write failed ({$cmd})");
        }

        while (true) {
            $line = @fgets($stream);
            if ($line === false) {
                $this->close();
                throw new SidecarException("sidecar closed or timed out awaiting the {$cmd} reply");
            }

            $reply = json_decode(trim($line), true);
            if (! is_array($reply)) {
                continue;
            }

            if (isset($reply['event'])) {
                $this->events[] = $reply;

                continue;
            }

            if (($reply['id'] ?? null) !== $id) {
                continue;
            }

            if (($reply['ok'] ?? false) !== true) {
                throw new SidecarException((string) ($reply['error'] ?? "sidecar error ({$cmd})"));
            }

            return $reply;
        }
    }

    /**
     * Notices (peer_connected / peer_left / …) received while waiting for
     * replies. Clears the buffer.
     *
     * @return list<array<string,mixed>>
     */
    public function drainEvents(): array
    {
        [$events, $this->events] = [$this->events, []];

        return $events;
    }

    public function close(): void
    {
        if (is_resource($this->stream)) {
            @fclose($this->stream);
        }
        $this->stream = null;
    }

    /** @return resource */
    private function stream()
    {
        if (is_resource($this->stream)) {
            return $this->stream;
        }

        $stream = @stream_socket_client(
            'unix://'.$this->socketPath,
            $errno,
            $error,
            $this->timeout,
        );

        if ($stream === false) {
            throw new SidecarException(
                "cannot connect to the SFU sidecar at {$this->socketPath}: ".($error ?: "errno {$errno}")
            );
        }

        $seconds = (int) floor($this->timeout);
        stream_set_timeout($stream, $seconds, (int) (($this->timeout - $seconds) * 1_000_000));

        return $this->stream = $stream;
    }
}