Download the PHP package event-stream/laravel without Composer

event-stream/laravel

Subscribable, resumable server-sent event (SSE) streams over Redis for Laravel.

Define a stream once on the server, publish typed messages to it from anywhere in your application, and let any number of browser clients subscribe to it over a single SSE connection. The server tracks a per-stream resume cursor, so a client that drops and reconnects picks up exactly where it left off — no duplicated and no missed messages. An optional state-stream layer keeps a JSON object graph automatically synced from the server to every client.

This is the backend package. The companion frontend package is event-stream-react. The two communicate over a small, versioned wire protocol and should be kept on matching major versions.

• Multi-stream multiplexing — one browser EventSource can subscribe to many streams at once; each is tracked and resumed independently.
• Resumable — every message carries a cursor; reconnects resume from the last delivered message using the standard Last-Event-ID mechanism.
• Authorization per stream — each stream decides who may subscribe.
• Interceptors — transform or suppress messages on their way to the client.
• Preludes — hand a freshly connected client a snapshot before live messages begin.
• State streams — mirror a server-side JSON graph to clients via granular set / delete / reset mutations instead of re-fetching.

Requirements

• PHP 8.2+
• Laravel 10, 11, or 12
• A Redis connection configured in config/database.php (the streams are backed by Redis Streams)
• A server setup that allows long-lived PHP responses (see Deployment notes)

Installation

The service provider is auto-discovered. Publish the config file:

This writes config/event-stream.php.

Quick start

This quick-start section is shared between the backend and frontend packages so both halves of the picture are in one place.

1. Backend — define a stream

A stream is a class. It declares a pattern for its key, decides who can subscribe, and exposes typed methods that publish messages.

2. Backend — register the stream

Add the class to config/event-stream.php:

3. Backend — expose the SSE route

Mount the bundled controller wherever you like, behind whatever middleware your app needs (the package intentionally does not register a route for you):

4. Backend — publish messages

From a job, controller, event listener — anywhere:

5. Frontend — wrap your app once

6. Frontend — subscribe

That is the whole round trip: publish on the server, receive in the browser, with resumable delivery handled for you.

State streams (synced graph)

For a server-owned object graph you want mirrored to clients without writing per-field event handling, use a state stream.

Backend reference

The rest of this document covers the backend in depth. For the frontend API, see the event-stream-react README.

How it works

1. A browser opens one EventSource to your SSE route, naming one or more streams via ?streams[]=<key> query parameters.
2. The controller resolves each key to a registered EventStream subclass, calls authorize() on each, and opens a single batched, blocking read across all of them against Redis.
3. As messages arrive they are run through each stream's interceptors and written to the client as SSE frames. Every live frame carries a composite id: (the Last-Event-ID) encoding a resume cursor per stream.
4. If the connection drops, the browser automatically reconnects and replays the last Last-Event-ID. The controller decodes it and resumes each stream from its own cursor.

Publishing is just an XADD to a Redis stream; subscribing is an XREAD. There is no broker process to run — Redis is the entire transport.

Defining streams

The stream pattern

Every subclass declares a streamPattern constant. Only { and } are special; everything else is a literal that must match exactly.

When a client subscribes to job.progress:abc-123, the registry finds the class whose pattern matches and instantiates it with that key. Extracted placeholder values are available as $this->args:

Construct instances with make(), passing placeholder values left to right:

Authorization

authorize() is required and runs once per subscription, with $this->args already populated. Return false to reject with a 403.

Publishing

Wrap send() in typed methods so call sites read clearly. The first argument is the message type (an arbitrary string your frontend switches on); the second is the JSON-serializable data payload. send() returns the Redis message id.

The published wire message is { type, data }; the client receives it as message.type and message.data (see Wire protocol).

Choosing where to start: initialCursor()

By default a new subscription only receives messages published after it connects ('$'). Override initialCursor() to replay history on first connect:

A reconnect with a Last-Event-ID always takes precedence over this — so initialCursor() only governs the very first connection.

Preludes: a snapshot on connect

Return a payload from preludeMessage() to hand every freshly connected client a one-off message before live messages begin. A prelude is server-synthesized (not read from Redis) and never advances the resume cursor.

On the client this arrives with meta.isPrelude === true.

Interceptors

Interceptors transform or suppress messages on their way to the client. Register them in registerInterceptors():

Return a (possibly modified) DeliveredStreamMessage to forward it, or null to drop it. Use withData() / withType() to produce modified copies. Interceptors run in registration order, each receiving the previous one's output. A suppressed message still advances the resume cursor, so it is never re-read.

State streams

StateStream extends EventStream to keep a JSON object graph synced to clients. The graph is stored flattened into a single Redis hash (one field per leaf, keyed by dot-notation path); every mutation also publishes a granular event describing exactly what changed, so the client patches its local copy instead of re-fetching.

Mutating the graph

Paths

Paths are dot-strings ('author.name') or segment arrays (['author', 'name']). Use the array form when a segment legitimately contains a dot; internally dots inside a segment are escaped so they round-trip. Setting a path first clears anything stored beneath it, so replacing a subtree never leaves stale leaves.

Container types: jsonPathTypes()

Because the graph is stored flat, an empty node has no inherent array-vs-object type. By default a node is rendered as an array only when its keys form a clean 0..n list. Override jsonPathTypes() to pin specific paths, using * as a single-segment wildcard:

The first matching pattern (in declaration order) wins. This map is shipped to the client in the prelude so it materializes empty containers with the same type.

Default state and TTL

The state hash key defaults to the stream key plus a :state suffix; override stateHashKey() to change it.

Consuming a stream from the server

Sometimes the server itself needs to read a stream (e.g. a queued job that waits for a particular message). Call connect():

Or block until a condition is met, with a total time budget:

Message expiration and pruning

Published messages stay in a stream's Redis log so a reconnecting client can resume from its Last-Event-ID. Redis streams have no per-message TTL, so nothing removes them on its own — the log grows until something trims it.

Each stream declares how long its messages live by overriding messageExpirationSeconds():

Returning null — the default — defers to event-stream.default_message_expiration_seconds (one hour). Set that config value to null to keep messages indefinitely unless a stream says otherwise.

Upgrading: if you published config/event-stream.php before this version, add default_message_expiration_seconds to your copy. The package normally backfills missing config keys, but not when the config is cached — which is the usual production setup. A missing key reads as null, so the prune treats every stream as "keep forever" and reports success having deleted nothing.

Because the method runs on a resolved instance, the window can depend on the stream's own placeholder values:

Keep this method defensive. It runs against every matching key the prune finds, including keys whose resource is long gone — a stream that throws here is logged and skipped, so its messages stop being pruned until the throw is fixed.

Running the prune

Expired messages are deleted by one command:

It walks every Redis stream key belonging to the app, resolves each to its registered stream class, and trims messages older than that stream's window. A stream left with nothing is deleted outright: Redis keeps an emptied stream key alive — unlike a list or a hash — so a stream key per resource would otherwise outlive every resource. Keys with no registered class behind them are left alone.

Each run reports what it did:

A stream whose window cannot be resolved — an ambiguous key, or a messageExpirationSeconds() that throws — is logged and skipped rather than ending the run, and the command says how many it skipped. The run still exits successfully; a rejected Redis command, which is not one stream's problem, fails it outright.

Nothing schedules this for you. Run it on whatever cadence suits the app — for most, from the scheduler in routes/console.php:

What a run costs

Trimming itself is cheap: XTRIM ... MINID against millisecond-timestamped stream ids, one script per stream. The cost is in getting there.

Every run walks the whole keyspace — SCAN visits every key the Redis database holds, matching and type-filtering as it goes, whether or not those keys are yours — and instantiates one stream object per key it does match, including whatever work that stream's messageExpirationSeconds() does. An app with a stream per resource and a messageExpirationSeconds() that queries the database is doing that query once per resource, per run.

The run's own output is the measure of this: in Pruned 1,204 messages from 37 streams (412 scanned), the gap between 412 and 37 is work spent on keys with nothing to prune. If that gap is wide and the scheduled cadence is tight, lengthen the cadence — expiration is a floor on how long messages live, not a promise they disappear the moment they expire.

One more thing worth knowing: the trim is exact, not approximate — no ~ or LIMIT. A first run against a stream that has gone unpruned for a long time deletes every expired entry in one synchronous step, which briefly blocks Redis in proportion to how many that is. Steady-state runs delete little and finish quickly; the first one after adopting this may not.

Expiration and reconnecting clients

A stream's expiration window is also the window in which a disconnected client can catch up. Resuming means handing back a Last-Event-ID, and if the messages after that id have been trimmed, XREAD simply returns what is left — the client receives newer messages with no indication that anything was missed.

The one-hour default is chosen with that in mind: it sits just above max_connection_seconds (55 minutes), so a client cycling through the normal forced reconnect always resumes well inside its stream's window. A client that goes away for longer — a closed laptop, a backgrounded tab, a phone offline in a tunnel — can come back to a gap.

Two things make this mostly theoretical, and one makes it real:

• StateStream is immune. Its prelude re-sends the entire graph on every connect, resume included, so a client that missed mutations is made whole regardless of what was trimmed.
• A stream left on the default initialCursor() of '$' never replays history anyway, so a first-time client loses nothing either way.
• A stream that overrides initialCursor() to '0-0' to replay its log does lose history — quietly. If a stream's log is the record rather than a catch-up buffer, give it a window long enough to hold everything worth replaying, or return null from messageExpirationSeconds() and let it keep everything.

Configuration

config/event-stream.php:

You may also pass an explicit list of stream classes to new EventStreamRegistry([...]) directly — handy in tests.

Wire protocol

The backend and event-stream-react share this contract. Keep the two packages on matching major versions.

Each SSE message frame's data: is JSON of the shape:

• The Last-Event-ID (and the lastEventId query fallback) is a composite cursor: streamKey1=redisId1;streamKey2=redisId2.
• Preludes carry messageId: null and never advance the cursor.
• State-stream mutations use reserved types: state_stream_mutation_reset, state_stream_mutation_set, state_stream_mutation_delete.

Deployment notes

An SSE response is a long-lived HTTP request. Make sure your stack allows it:

• Workers — each open connection holds one PHP worker (php-fpm/Octane) for its lifetime. Size your worker pool for the number of concurrent subscribers.
• Buffering — the controller sends X-Accel-Buffering: no and disables output buffering. Ensure no proxy (nginx, load balancer) re-enables response buffering on this route.
• Timeouts — raise proxy read timeouts above max_connection_seconds, or rely on the built-in periodic reconnect. PHP's own max_execution_time is handled for you: the controller calls set_time_limit() to cover the connection's lifetime, so no ini change is needed.
• Redis — the reader uses blocking XREAD; ensure your Redis client/proxy read timeout exceeds heartbeat_interval_ms.

Publishing new versions

See PUBLISHING.md.

License

MIT