reactphp-kernel/src/Ipc/SocketPairIpc.php

<?php

declare(strict_types=1);

namespace Blax\ReactPhpKernel\Ipc;

use React\EventLoop\LoopInterface;

/**
 * Event-driven IPC using Unix socket pairs.
 *
 * Provides instant notification when a child process sends data, eliminating
 * polling. Lifted verbatim (bar the namespace) from blax-software/laravel-websockets
 * because it is already framework-agnostic — it belongs to the shared backbone.
 *
 * Usage:
 *   1. Before fork:            $ipc = SocketPairIpc::create($loop);
 *   2. After fork in parent:   $ipc->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');
    }
}