Cross-node Pub/Sub & Streams

// app.php — before $app->run()
use ZealPHP\Store;
use ZealPHP\App;

// Step 1: tell Store to use Redis. ZEALPHP_STORE_BACKEND=redis env works too.
Store::defaultBackend(Store::BACKEND_REDIS);

// Step 2: register a subscriber at boot. Runs in EVERY worker.
App::subscribe('hello:world', function (string $payload, string $channel) {
    error_log("[$channel] received: $payload");
});

// Step 3: publish from anywhere — a route, a timer, a CLI script, …
$receivers = Store::publish('hello:world', 'hi from anywhere');
// Returns number of receivers — across ALL workers AND ALL connected servers.

App::subscribeReliable('orders', function (string $payload, string $id, string $stream): bool {
    $order = json_decode($payload, true);
    $ok = processOrder($order);
    return $ok;  // true → XACK (done); false/throw → leave pending, retried on reconnect
});

// Anywhere:
$messageId = Store::publishReliable('orders', json_encode($order));
// Returns the Redis message ID, e.g. '1779520329297-0' — durable when AOF/RDB is on.

// app.php — boot
Store::defaultBackend(Store::BACKEND_REDIS);
$myServerId = gethostname() . ':' . getmypid();

// Shared mapping: which server owns each connected client.
Store::make('ws_owner', 4096, ['server' => [Store::TYPE_STRING, 64]]);

// Per-worker local: client_id → local fd (only valid in THIS process)
$localFds = [];

App::ws('/ws', function ($server, $frame) use (&$localFds) {
    // Process inbound from local clients normally.
});

// Each server's subscriber only handles its OWN routed messages.
App::subscribe("ws:server:$myServerId", function (string $payload) use ($server, &$localFds) {
    $msg = json_decode($payload, true);
    $fd  = $localFds[$msg['client_id']] ?? null;
    if ($fd !== null && $server->isEstablished($fd)) {
        $server->push($fd, $msg['data']);
    }
});

// Anywhere: route a message to client X, regardless of which server holds it.
function sendToClient(string $clientId, string $data): void {
    $owner = Store::get('ws_owner', $clientId, 'server');
    if ($owner === null) { return; } // client not connected anywhere
    Store::publish("ws:server:$owner", json_encode([
        'client_id' => $clientId,
        'data'      => $data,
    ]));
}