RedisStreams

in package

• ZealPHP

Per-worker Redis Streams consumer runner.

Table of Contents

Constants

DEFAULT_RECLAIM_EVERY_SEC  : mixed = 30

Default wall-time cadence (seconds) between orphan-reclaim passes.

DEFAULT_RECLAIM_MIN_IDLE_MS  : mixed = 60000

Default minimum idle time (ms) before a pending entry is eligible for reclaim.

RECLAIM_MAX_ITERATIONS  : mixed = 10000

Max XAUTOCLAIM cursor iterations per reclaim pass (drain-loop safety cap).

Properties

$ackPool  : RedisConnectionPool|null

Small dedicated pool of clients for XACK. Each ACKed message used to spin up a BRAND-NEW connection (connect + close per message), so a busy stream opened a TCP connection per delivery — a connection storm under load. A size-2 pool reuses connections across messages while still giving each dispatch coroutine a private socket (XACK from two cors on one socket would interleave RESP frames). Built lazily in the runner coroutine; closed by stop().

$consumerName  : string

$consumers  : array<int, array{stream: string, group: string, handler: callable, blockMs: int, batchSize: int}>

$opts  : array<string|int, mixed>

$reclaimCount  : int

Per-pass XAUTOCLAIM batch size (entries claimed per cursor iteration).

$reclaimEverySec  : int

Wall-time cadence (seconds) between orphan-reclaim passes.

$reclaimMinIdleMs  : int

Minimum idle time (ms) before a pending entry is reclaimed.

$running  : Atomic

$url  : string

Methods

__construct()  : mixed

consumerName()  : string

consumers()  : array<int, array{stream: string, group: string, handler: callable, blockMs: int, batchSize: int}>

isRunning()  : bool

reclaimEverySec()  : int

Configured wall-time cadence (seconds) between orphan-reclaim passes.

reclaimMinIdleMs()  : int

Configured minimum pending idle (ms) before a message is reclaimed.

reclaimPolicy()  : self

Tune the orphan-reclaim cadence + min-idle without re-constructing.

register()  : void

Register a stream + group + handler. Handler signature: function (string $payload, string $messageId, string $stream): bool Return true to XACK (message removed from pending). Return false OR throw to leave pending (retried on consumer recovery).

start()  : void

Spawn the runner coroutine. MUST be called from inside a coroutine context.

stop()  : void

Signal the runner to exit cleanly at the next BLOCK timeout.

ackPool()  : RedisConnectionPool

Lazily-built ACK connection pool (size 2). Constructing it is cheap — the pool only connects on first acquire(), which happens inside the dispatch coroutine. Reused across messages so a busy stream doesn't connect-per-XACK.

atomicIsZero()  : bool

backoffSeconds()  : float

dispatch()  : void

ensureGroups()  : void

reclaimAll()  : void

Run one orphan-reclaim pass across every registered consumer. Each consumer's pending list is XAUTOCLAIM-iterated (cursor walked to '0-0') and every reclaimed entry is re-dispatched through the same handler + XACK-on-success path as a fresh read.

reclaimPass()  : void

Drive the XAUTOCLAIM cursor for ONE consumer until it returns to '0-0' (end of the pending list), dispatching each reclaimed entry. Both the claim mechanism and the per-entry dispatch are injected so the cursor-iteration logic is unit-testable without a live RedisClient (the runner injects $client->xautoclaim() + the real dispatch(); a test injects a fake page source + a synchronous dispatch recorder).

runHandlerDecision()  : bool

Pure handler-decision core shared by read + reclaim dispatch. Runs the handler; on a strict true return it invokes $ack (the XACK action); a non-true return OR a thrown handler leaves the entry pending (no XACK) and is swallowed so the runner survives. Returns the decision (true = ACKed) for testability — the caller's $ack does the real network XACK on a private pooled socket.

runner()  : void

Group consumers by (group, blockMs, batchSize) so one XREADGROUP call can fetch from multiple streams sharing the same consumer-group params.