Alpha ZealPHP is early-stage and under active development. APIs may change between minor versions until v1.0. Feedback and bug reports welcome on GitHub.
Esc
API Index — Namespaces, Packages, Reports, Indices

RedisPubSub
in package

FinalYes

Per-worker pub/sub runner.

Owns ONE dedicated Redis connection (separate from the pool — SUBSCRIBE monopolises a connection) and one runner coroutine. Routes inbound messages to registered handlers via go() per message so a slow handler can't block the next read.

Wake-up + clean shutdown: a private sentinel channel is subscribed alongside the user channels; stop() publishes a marker to it via a separate client. The subscribe consumer detects the marker, throws PubSubStopException, the driver catches it, the runner exits.

Reconnect: connection drop while subscribed → bounded exponential backoff (capped at 5 s) → re-connect → re-SUBSCRIBE / re-PSUBSCRIBE everything. Messages published during the reconnect window are lost (Redis pub/sub has no buffering) — use Store::publishReliable for at-least-once.

Table of Contents

Properties

$exactHandlers  : array<string, array<int, callable>>
Exact-channel handlers keyed by channel name.
$maxAttempts  : int
$opts  : array<string|int, mixed>
$patternHandlers  : array<string, array<int, callable>>
Pattern handlers keyed by PSUBSCRIBE pattern (contains *).
$prefix  : string
$running  : Atomic
Atomic flag (1 = running, 0 = stopped). Cross-coroutine visibility via OpenSwoole\Atomic.
$stats  : Stats
Per-worker counters exposed via stats().
$stopChannel  : string
Private sentinel channel used by stop() to wake the subscriber loop cleanly.
$url  : string

Methods

__construct()  : mixed
exactChannels()  : array<int, string>
Return the exact channel names with registered handlers.
isRunning()  : bool
Return true when the runner coroutine is active.
patternChannels()  : array<int, string>
Return the PSUBSCRIBE pattern strings with registered handlers.
register()  : void
Register a handler. Channels containing '*' are PSUBSCRIBE patterns; everything else is SUBSCRIBE exact. Multiple handlers per channel allowed; all fire on each message.
start()  : void
Spawn the runner coroutine. Idempotent — re-calling while already running is a no-op. MUST be called from inside a coroutine context.
stats()  : Stats
Per-worker stats — pubsub_reconnects_total, pubsub_messages_received_total, pubsub_handler_errors_total.
stop()  : void
Signal the runner to exit cleanly. Publishes a sentinel to the private stop channel via a NEW client (the running subscriber owns its connection, can't publish to itself).
stopChannel()  : string
Return the private stop-sentinel channel name (used by stop() internally).
atomicIsZero()  : bool
Read $a->get() === 0 via an out-of-line method.
backoffSeconds()  : float
Compute the reconnect wait time for a given attempt number.
dispatch()  : void
Fan out an inbound message to all matching handlers via go().
runner()  : void
Main subscriber loop running inside its own coroutine.

Properties

$exactHandlers

Exact-channel handlers keyed by channel name.

private array<string, array<int, callable>> $exactHandlers = []

$patternHandlers

Pattern handlers keyed by PSUBSCRIBE pattern (contains *).

private array<string, array<int, callable>> $patternHandlers = []

$running

Atomic flag (1 = running, 0 = stopped). Cross-coroutine visibility via OpenSwoole\Atomic.

private Atomic $running

$stopChannel

Private sentinel channel used by stop() to wake the subscriber loop cleanly.

private string $stopChannel

Methods

__construct()

public __construct(string $url[, string $prefix = 'zealstore' ][, int $maxAttempts = 0 ][, array{prefer?: "auto"|"phpredis"|"predis"} $opts = [] ]) : mixed
Parameters
$url : string
$prefix : string = 'zealstore'
$maxAttempts : int = 0

Bounded reconnect attempts (0 = unlimited, the default; preserves pre-H10 behaviour). When set to N > 0, the runner gives up after N consecutive backoff cycles and logs a final error. Use this when "keep trying forever" isn't acceptable — e.g. a CI worker that should fail loudly if its Redis disappears.

$opts : array{prefer?: "auto"|"phpredis"|"predis"} = []

Driver preference for the runner's connections. Forced to ['prefer' => 'predis'] by App::wirePubSubBoot() when phpredis would block the worker under HOOK_ALL=0 (H7) — predis SUBSCRIBE yields on its pure-PHP socket regardless of HOOK_ALL.

exactChannels()

Return the exact channel names with registered handlers.

public exactChannels() : array<int, string>
Return values
array<int, string>

isRunning()

Return true when the runner coroutine is active.

public isRunning() : bool
Return values
bool

patternChannels()

Return the PSUBSCRIBE pattern strings with registered handlers.

public patternChannels() : array<int, string>
Return values
array<int, string>

register()

Register a handler. Channels containing '*' are PSUBSCRIBE patterns; everything else is SUBSCRIBE exact. Multiple handlers per channel allowed; all fire on each message.

public register(string $channelOrPattern, callable $handler) : void
Parameters
$channelOrPattern : string
$handler : callable

start()

Spawn the runner coroutine. Idempotent — re-calling while already running is a no-op. MUST be called from inside a coroutine context.

public start() : void

stats()

Per-worker stats — pubsub_reconnects_total, pubsub_messages_received_total, pubsub_handler_errors_total.

public stats() : Stats
Return values
Stats

stop()

Signal the runner to exit cleanly. Publishes a sentinel to the private stop channel via a NEW client (the running subscriber owns its connection, can't publish to itself).

public stop() : void

stopChannel()

Return the private stop-sentinel channel name (used by stop() internally).

public stopChannel() : string
Return values
string

atomicIsZero()

Read $a->get() === 0 via an out-of-line method.

private static atomicIsZero(Atomic $a) : bool

An indirect read prevents PHPStan from constant-folding the while ($this->running->get() === 1) loop condition when it can prove running was set to 1 just above — which would eliminate the loop body entirely under strict flow analysis.

Parameters
$a : Atomic
Return values
bool

backoffSeconds()

Compute the reconnect wait time for a given attempt number.

private static backoffSeconds(int $attempt) : float

Uses bounded exponential backoff: 0.1 × 2^attempt seconds, capped at 5.0 s. Sequence: 0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0, 5.0, …

Parameters
$attempt : int
Return values
float

dispatch()

Fan out an inbound message to all matching handlers via go().

private dispatch(string $payload, string $channel, string|null $pattern) : void

Each handler runs in its own coroutine so a slow handler cannot block the next message read. Exceptions thrown by handlers are caught, counted in pubsub_handler_errors_total, and logged via error_log().

Parameters
$payload : string
$channel : string
$pattern : string|null

runner()

Main subscriber loop running inside its own coroutine.

private runner() : void

Connects a dedicated RedisClient, subscribes to all registered exact channels + PSUBSCRIBE patterns + the private $stopChannel, then reads messages in a blocking loop. On PubSubStopException (sentinel received via stop()) the loop exits cleanly. On StoreException (connection drop) it applies bounded exponential backoff and reconnects, up to $maxAttempts times (0 = unlimited).

On this page