From 96b3de6a535ea5a57a5073321c54742e4cecb409 Mon Sep 17 00:00:00 2001 From: Blax Software Date: Tue, 7 Jul 2026 13:07:47 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20initial=20reactphp-kernel=20=E2=80=94?= =?UTF-8?q?=20shared=20ReactPHP=20backbone?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Protocol-agnostic backbone extracted from blax-software/laravel-websockets so WebSockets, WebRTC signaling and future protocols attach to one long-lived process + loop + IPC + supervision: - Kernel: loop lifecycle, server registration, signals, graceful stop - Contracts\Server: attachable protocol server (boot/shutdown) - Server\SocketServerFactory: plain/TLS listening socket (no framework coupling) - Ipc\SocketPairIpc: event-driven parent/child IPC (lifted verbatim) - Process\SignalHandler + ChildReaper (reusable form of laravel-websockets #982) rel learn-atc #1055 #1051 Co-Authored-By: Claude Opus 4.8 --- .gitignore | 8 ++ LICENSE | 21 ++++ README.md | 79 +++++++++++++++ composer.json | 41 ++++++++ src/Contracts/Server.php | 34 +++++++ src/Ipc/SocketPairIpc.php | 154 +++++++++++++++++++++++++++++ src/Kernel.php | 147 +++++++++++++++++++++++++++ src/Process/ChildReaper.php | 81 +++++++++++++++ src/Process/SignalHandler.php | 33 +++++++ src/Server/SocketServerFactory.php | 31 ++++++ 10 files changed, 629 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 composer.json create mode 100644 src/Contracts/Server.php create mode 100644 src/Ipc/SocketPairIpc.php create mode 100644 src/Kernel.php create mode 100644 src/Process/ChildReaper.php create mode 100644 src/Process/SignalHandler.php create mode 100644 src/Server/SocketServerFactory.php diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1e795c7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +/vendor/ +composer.lock +.phpunit.result.cache +.phpunit.cache/ +*.log +.DS_Store +.idea/ +.vscode/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..9d3a4f5 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Blax Software + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..937d2e2 --- /dev/null +++ b/README.md @@ -0,0 +1,79 @@ +# blax-software/reactphp-kernel + +A small, protocol-agnostic **ReactPHP backbone**: one long-lived process, one +event loop, one IPC primitive, and one signal/graceful-shutdown story that many +protocol servers attach to. + +It is the shared foundation extracted from +[`blax-software/laravel-websockets`](https://git.blax.at/blax-software/laravel-websockets) +so that WebSockets, **WebRTC signaling/media** +([`blax-software/laravel-webrtc`](https://git.blax.at/blax-software/laravel-webrtc)), +and future protocols (SIP, ...) do not each re-implement the plumbing. + +``` +blax-software/reactphp-kernel <- this package (the backbone) + ├─ laravel-websockets <- WS on the kernel + └─ laravel-webrtc <- WebRTC signaling on the kernel + └─ Rust/str0m media core <- via ext-php-rs +``` + +## What's in the box + +| Piece | Responsibility | +|---|---| +| `Kernel` | Owns the loop; registers `Server`s; installs signals; `run()` / graceful `stop()`. | +| `Contracts\Server` | An attachable protocol server: `boot(LoopInterface)` (non-blocking) + `shutdown()`. | +| `Server\SocketServerFactory` | Plain-TCP or TLS listening socket (no framework coupling). | +| `Ipc\SocketPairIpc` | Event-driven parent/child IPC over a Unix socket pair (no polling). | +| `Process\SignalHandler` | SIGINT/SIGTERM → graceful shutdown (no-op without ext-pcntl). | +| `Process\ChildReaper` | Reaps exited forked children (SIGCHLD + periodic backstop) — the reusable form of laravel-websockets #982. | + +## Usage + +```php +use BlaxSoftware\ReactPhpKernel\Kernel; + +(new Kernel()) + ->reapChildren() // optional: auto-reap forked children + ->register($webSocketServer) // any Contracts\Server + ->register($webRtcSignaling) + ->onBoot(fn (Kernel $k) => /* warm caches, announce ready, ... */ null) + ->run(); // boots servers, installs signals, runs the loop (blocks) +``` + +Implement `Contracts\Server` to plug in a protocol: + +```php +use BlaxSoftware\ReactPhpKernel\Contracts\Server; +use BlaxSoftware\ReactPhpKernel\Server\SocketServerFactory; +use React\EventLoop\LoopInterface; + +final class MyServer implements Server +{ + private $socket = null; + + public function name(): string { return 'my-server'; } + + public function boot(LoopInterface $loop): void + { + $this->socket = SocketServerFactory::create('0.0.0.0:9000', $loop); + $this->socket->on('connection', function ($conn) { /* ... */ }); + } + + public function shutdown(): void + { + $this->socket?->close(); + } +} +``` + +## Requirements + +- PHP >= 8.1 +- `react/event-loop`, `react/socket`, `react/stream` +- `ext-pcntl` (suggested) for signal handling + child reaping +- `ext-sockets` (suggested) for `SocketPairIpc` + +## License + +MIT © Blax Software diff --git a/composer.json b/composer.json new file mode 100644 index 0000000..4199004 --- /dev/null +++ b/composer.json @@ -0,0 +1,41 @@ +{ + "name": "blax-software/reactphp-kernel", + "description": "Protocol-agnostic ReactPHP backbone: event-loop lifecycle, socket accept, IPC, signal handling and graceful shutdown, shared by blax realtime servers (WebSockets, WebRTC, ...).", + "type": "library", + "license": "MIT", + "keywords": ["reactphp", "event-loop", "kernel", "realtime", "websocket", "webrtc", "ipc", "daemon"], + "authors": [ + { + "name": "Blax Software", + "email": "office@blax.at" + } + ], + "require": { + "php": ">=8.1", + "react/event-loop": "^1.5", + "react/socket": "^1.15", + "react/stream": "^1.3" + }, + "require-dev": { + "phpunit/phpunit": "^10.5" + }, + "suggest": { + "ext-pcntl": "Signal handling (graceful shutdown) and reaping of forked children.", + "ext-sockets": "Required for the SocketPairIpc parent<->child channel." + }, + "autoload": { + "psr-4": { + "BlaxSoftware\\ReactPhpKernel\\": "src/" + } + }, + "autoload-dev": { + "psr-4": { + "BlaxSoftware\\ReactPhpKernel\\Tests\\": "tests/" + } + }, + "config": { + "sort-packages": true + }, + "minimum-stability": "stable", + "prefer-stable": true +} diff --git a/src/Contracts/Server.php b/src/Contracts/Server.php new file mode 100644 index 0000000..145adfd --- /dev/null +++ b/src/Contracts/Server.php @@ -0,0 +1,34 @@ +run() itself — it only + * registers listeners on the loop the Kernel passes in. The Kernel calls run(). + */ +interface Server +{ + /** Human-readable identifier, used in logs / diagnostics. */ + public function name(): string; + + /** + * Attach to the shared loop: bind listening sockets, register read streams, + * periodic timers, etc. Returns immediately (non-blocking). + */ + public function boot(LoopInterface $loop): void; + + /** + * Gracefully tear down: stop accepting, close listeners, drain/close open + * connections. Called by the Kernel on SIGINT/SIGTERM before the loop stops. + */ + public function shutdown(): void; +} diff --git a/src/Ipc/SocketPairIpc.php b/src/Ipc/SocketPairIpc.php new file mode 100644 index 0000000..5cb8a65 --- /dev/null +++ b/src/Ipc/SocketPairIpc.php @@ -0,0 +1,154 @@ +setupParent($onData, $onClose); + * 3. After fork in child: $ipc->setupChild(); $ipc->sendToParent($data); + */ +final class SocketPairIpc +{ + /** + * Socket pair: [0] = parent side, [1] = child side. + * + * @var resource[]|null + */ + private ?array $sockets = null; + + private ?LoopInterface $loop = null; + + /** 'parent' | 'child' | null (unconfigured). */ + private ?string $role = null; + + /** Exported stream resource for ReactPHP. */ + private $stream = null; + + /** Create a new socket pair for IPC. Must be called BEFORE fork(). */ + public static function create(LoopInterface $loop): self + { + $instance = new self(); + $instance->loop = $loop; + + if (! socket_create_pair(AF_UNIX, SOCK_STREAM, 0, $sockets)) { + throw new \RuntimeException('Failed to create socket pair: ' . socket_strerror(socket_last_error())); + } + + $instance->sockets = $sockets; + + return $instance; + } + + /** + * Setup the parent side after fork: close the child socket and read async. + * + * @param callable $onData function(string $message): void — newline-delimited frames + * @param callable|null $onClose function(): void — child closed the channel + */ + public function setupParent(callable $onData, ?callable $onClose = null): void + { + if ($this->role !== null) { + throw new \LogicException('IPC already configured'); + } + + $this->role = 'parent'; + + socket_close($this->sockets[1]); + socket_set_nonblock($this->sockets[0]); + + $this->stream = socket_export_stream($this->sockets[0]); + if ($this->stream === false) { + throw new \RuntimeException('Failed to export socket as stream'); + } + + $buffer = ''; + + $this->loop->addReadStream($this->stream, function ($stream) use ($onData, $onClose, &$buffer) { + $data = @fread($stream, 65536); + + if ($data === false || $data === '') { + if ($buffer !== '') { + $onData($buffer); + } + $this->loop->removeReadStream($stream); + fclose($stream); + if ($onClose) { + $onClose(); + } + + return; + } + + $buffer .= $data; + + while (($pos = strpos($buffer, "\n")) !== false) { + $message = substr($buffer, 0, $pos); + $buffer = substr($buffer, $pos + 1); + + if ($message !== '') { + $onData($message); + } + } + }); + } + + /** Setup the child side after fork: close the parent socket. */ + public function setupChild(): void + { + if ($this->role !== null) { + throw new \LogicException('IPC already configured'); + } + + $this->role = 'child'; + + socket_close($this->sockets[0]); + } + + /** + * Send data from child to parent (newline-delimited; do not embed newlines). + * Call only from the child after setupChild(). + */ + public function sendToParent(string $data): bool + { + if ($this->role !== 'child') { + throw new \LogicException('sendToParent can only be called from child'); + } + + $message = $data . "\n"; + $written = socket_write($this->sockets[1], $message, strlen($message)); + + return $written === strlen($message); + } + + /** Close the child socket (call at the end of the child process). */ + public function closeChild(): void + { + if ($this->role === 'child' && $this->sockets[1]) { + socket_close($this->sockets[1]); + } + } + + /** @return resource */ + public function getChildSocket() + { + return $this->sockets[1]; + } + + /** Whether this system supports socket-pair IPC. */ + public static function isSupported(): bool + { + return function_exists('socket_create_pair') + && function_exists('socket_export_stream'); + } +} diff --git a/src/Kernel.php b/src/Kernel.php new file mode 100644 index 0000000..3973098 --- /dev/null +++ b/src/Kernel.php @@ -0,0 +1,147 @@ +reapChildren() // optional: auto-reap forked children + * ->register($webSocketServer) // any Contracts\Server + * ->register($webRtcSignaling) + * ->onBoot(fn (Kernel $k) => ...) + * ->run(); // boots servers, installs signals, runs the loop + */ +final class Kernel +{ + private LoopInterface $loop; + + /** @var Server[] */ + private array $servers = []; + + /** @var array */ + private array $onBoot = []; + + /** @var array */ + private array $onShutdown = []; + + private bool $running = false; + + private ?ChildReaper $reaper = null; + + public function __construct(?LoopInterface $loop = null) + { + // Default to the global loop so co-located libraries share one reactor. + $this->loop = $loop ?? Loop::get(); + } + + public function loop(): LoopInterface + { + return $this->loop; + } + + /** Attach a protocol server (WS, WebRTC, ...) to the shared loop. */ + public function register(Server $server): self + { + $this->servers[] = $server; + + return $this; + } + + public function onBoot(callable $callback): self + { + $this->onBoot[] = $callback; + + return $this; + } + + public function onShutdown(callable $callback): self + { + $this->onShutdown[] = $callback; + + return $this; + } + + /** + * Automatically reap exited forked children so they never accumulate as + * zombies (the concern behind laravel-websockets #982). Opt-in because not + * every deployment forks. + */ + public function reapChildren(float $interval = 1.0): self + { + $this->reaper = new ChildReaper($this->loop); + $this->reaper->install($interval); + + return $this; + } + + /** Boot every registered server, install signal handlers, run the loop (blocks). */ + public function run(): void + { + if ($this->running) { + return; + } + $this->running = true; + + foreach ($this->servers as $server) { + $server->boot($this->loop); + } + + foreach ($this->onBoot as $callback) { + $callback($this); + } + + SignalHandler::install($this->loop, fn () => $this->stop()); + + $this->loop->run(); + } + + /** Graceful shutdown: tear down servers, run shutdown hooks, stop the loop. */ + public function stop(): void + { + if (! $this->running) { + return; + } + + foreach ($this->servers as $server) { + try { + $server->shutdown(); + } catch (\Throwable) { + // A failing teardown must not block the others. + } + } + + foreach ($this->onShutdown as $callback) { + try { + $callback($this); + } catch (\Throwable) { + // ignore + } + } + + $this->reaper?->reapAll(); + $this->reaper?->uninstall(); + + $this->running = false; + $this->loop->stop(); + } + + public function isRunning(): bool + { + return $this->running; + } +} diff --git a/src/Process/ChildReaper.php b/src/Process/ChildReaper.php new file mode 100644 index 0000000..0648ef3 --- /dev/null +++ b/src/Process/ChildReaper.php @@ -0,0 +1,81 @@ +loop = $loop; + } + + public function install(float $interval = 1.0): void + { + if (! \function_exists('pcntl_waitpid')) { + return; // No pcntl — nothing to reap (this build cannot fork). + } + + if (\defined('SIGCHLD')) { + $this->signalHandler = fn () => $this->reapAll(); + try { + $this->loop->addSignal(SIGCHLD, $this->signalHandler); + } catch (\BadMethodCallException) { + $this->signalHandler = null; // loop has no signal support + } + } + + $this->timer = $this->loop->addPeriodicTimer($interval, fn () => $this->reapAll()); + } + + /** Reap every child that has already exited. Returns how many were reaped. */ + public function reapAll(): int + { + if (! \function_exists('pcntl_waitpid')) { + return 0; + } + + $reaped = 0; + $status = 0; + while (($pid = \pcntl_waitpid(-1, $status, WNOHANG)) > 0) { + $reaped++; + } + + return $reaped; + } + + public function uninstall(): void + { + if ($this->timer !== null) { + $this->loop->cancelTimer($this->timer); + $this->timer = null; + } + + if ($this->signalHandler !== null && \defined('SIGCHLD')) { + try { + $this->loop->removeSignal(SIGCHLD, $this->signalHandler); + } catch (\Throwable) { + // ignore + } + $this->signalHandler = null; + } + } +} diff --git a/src/Process/SignalHandler.php b/src/Process/SignalHandler.php new file mode 100644 index 0000000..e7b653c --- /dev/null +++ b/src/Process/SignalHandler.php @@ -0,0 +1,33 @@ +addSignal(SIGINT, $handler); + $loop->addSignal(SIGTERM, $handler); + } catch (\BadMethodCallException) { + // The active loop provides no signal support; ignore. + } + } +} diff --git a/src/Server/SocketServerFactory.php b/src/Server/SocketServerFactory.php new file mode 100644 index 0000000..69b62f3 --- /dev/null +++ b/src/Server/SocketServerFactory.php @@ -0,0 +1,31 @@ + $tls ReactPHP TLS context (e.g. ['local_cert' => ..., 'local_pk' => ...]); [] = plain TCP + */ + public static function create(string $address, LoopInterface $loop, array $tls = []): SocketServer + { + if ($tls !== []) { + return new SocketServer("tls://{$address}", ['tls' => $tls], $loop); + } + + return new SocketServer($address, [], $loop); + } +}