event-stream-react

React client for subscribable, resumable server-sent event (SSE) streams over Redis.

Subscribe a component to one or more server-defined streams and receive messages as they are published. The client multiplexes every subscription onto a single browser EventSource, tracks a resume cursor per stream, and transparently reconnects — so a dropped connection resumes exactly where it left off with no duplicated or missed messages. An optional state-stream layer mirrors a server-owned JSON object graph into an observable you can render directly.

This is the frontend package. The companion backend package is event-stream/laravel (Laravel + Redis). The two communicate over a small, versioned wire protocol and should be kept on matching major versions.

• One connection, many streams — subscriptions are deduplicated onto a single EventSource; each stream is tracked and resumed independently.
• Resumable — reconnects replay from the last delivered message via the standard Last-Event-ID mechanism, managed for you.
• Hook-first — useSubscribeToStreams for raw messages; useStreamStateObject for a synced object graph.
• Synced state graphs — render a server-owned graph from an observable that patches itself from granular set / delete / reset mutations.

Requirements & installation

npm install event-stream-react

This package has three peer dependencies you install alongside it:

npm install react mobx react-state-object

| Peer | Why it must be a peer (not bundled) | | --- | --- | | react | A single React instance must back your whole tree. | | mobx | The synced-graph layer exposes MobX observables; reactivity only works when your app and this package share one MobX runtime. | | react-state-object | Provides the observable "state object" base class and the dependency-injection context this package builds on (see below). |

What is react-state-object? It is a small library for observable, class-based state objects in React. A state object is a class whose fields are MobX observables; you mount an instance for a component's lifetime with a hook, and components re-render when the fields they read change. It also ships a lightweight dependency-injection context — a provider binds an instance, and descendants resolve it by class. This package uses both pieces: the app-wide connection manager is a state object provided through that context, and the synced graph (StreamStateObject) is a state object you mount per component. You rarely call react-state-object directly, but it must be installed and must be the same copy your app uses.

Quick start

This quick-start section is shared between the frontend and backend 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.

// app/Streams/JobProgressStream.php
namespace App\Streams;

use EventStream\EventStream;
use Illuminate\Http\Request;

class JobProgressStream extends EventStream
{
    // {jobId} is a placeholder; a key like "job.progress:abc-123"
    // makes $this->args->jobId === "abc-123".
    const streamPattern = 'job.progress:{jobId}';

    public function authorize(Request $request): bool
    {
        // Return true to allow the subscription, false for a 403.
        return $request->user() !== null;
    }

    // A typed publish method. `send()` returns the new Redis message id.
    public function sendProgress(int $percent): string
    {
        return $this->send('progress', ['percent' => $percent]);
    }
}

2. Backend — register the stream

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

'streams' => [
    App\Streams\JobProgressStream::class,
],

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):

// routes/web.php
use EventStream\Http\StreamSubscriptionController;

Route::get('/streams', [StreamSubscriptionController::class, 'stream'])
    ->middleware(['web', 'auth'])
    ->name('streams.subscribe');

4. Backend — publish messages

From a job, controller, event listener — anywhere:

use App\Streams\JobProgressStream;

JobProgressStream::make('abc-123')->sendProgress(50);

5. Frontend — wrap your app once

import { EventStreamProvider } from 'event-stream-react';

function Root() {
  return (
    // `url` must match the route you mounted the controller on.
    <EventStreamProvider url="/streams">
      <App />
    </EventStreamProvider>
  );
}

6. Frontend — subscribe

import { useState } from 'react';
import { useSubscribeToStreams } from 'event-stream-react';

function JobProgress({ jobId }: { jobId: string }) {
  const [percent, setPercent] = useState(0);

  useSubscribeToStreams(
    `job.progress:${jobId}`,
    ({ message }) => {
      if (message.type === 'progress') {
        setPercent((message.data as { percent: number }).percent);
      }
    }
  );

  return <div>{percent}% complete</div>;
}

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
namespace App\Streams;

use EventStream\StateStream;
use Illuminate\Http\Request;

class DocumentStateStream extends StateStream
{
    const streamPattern = 'document:{documentId}';

    public function authorize(Request $request): bool
    {
        return $request->user() !== null;
    }
}

// Mutate the graph anywhere — each call publishes a granular mutation:
$doc = DocumentStateStream::make('abc-123');
$doc->setState('title', 'Quarterly report');
$doc->setState('author.name', 'Ada');
$doc->arrayAppend('comments', ['body' => 'Looks good']);

// Frontend — `state` stays current automatically.
import { observer } from 'mobx-react-lite';
import { useStreamStateObject } from 'event-stream-react';

const Document = observer(({ id }: { id: string }) => {
  const doc = useStreamStateObject(`document:${id}`);
  return <h1>{String(doc.state.title ?? '')}</h1>;
});

Frontend reference

The rest of this document covers the frontend in depth. For defining streams, publishing, authorization, interceptors, and the state-stream server API, see the event-stream/laravel README.

EventStreamProvider

Mount this once near the root of your app. It creates the single app-wide connection manager and makes it available to every hook below.

<EventStreamProvider url="/streams" withCredentials>
  <App />
</EventStreamProvider>

| Prop | Type | Default | Notes | | --- | --- | --- | --- | | url | string | '/streams' | Base URL of the SSE endpoint. Absolute or relative (relative is resolved against the page origin). Must match where you mounted the backend controller. | | withCredentials | boolean | true | Whether the EventSource sends cookies. Keep true for cookie/session-authenticated routes. |

Everything below must be rendered inside this provider.

useSubscribeToStreams

Subscribe a component to one or more streams. The callback fires for every message delivered to those streams while the component is mounted.

useSubscribeToStreams(streamKeysOrKey, (envelope) => { ... });

• First argument — a stream key, or an array of keys, or stream objects (see Resuming from a known position). The subscription is re-created only when the set of keys changes, so an inline callback is fine — it never causes a reconnect.
• Callback — receives a StreamEnvelope:

type StreamEnvelope = {
  meta: {
    streamKey: string;        // which stream this message belongs to
    messageId: string | null; // Redis id, or null for a prelude
    isPrelude: boolean;       // true for the on-connect snapshot
  };
  message: {
    type: string;             // the type your backend sent
    data: unknown;            // the payload your backend sent
  };
};

Switch on message.type, and check meta.isPrelude if a stream sends an on-connect snapshot you want to handle specially:

useSubscribeToStreams(
  [`job.progress:${jobId}`, `job.logs:${jobId}`],
  ({ meta, message }) => {
    if (meta.isPrelude) {
      // one-off snapshot the stream emits on connect
      return;
    }
    switch (message.type) {
      case 'progress':
        setPercent((message.data as { percent: number }).percent);
        break;
      case 'completed':
        setDone(true);
        break;
    }
  }
);

Resuming from a known position

Pass a stream object instead of a bare key to start from a specific message. latestMessageId is a Redis stream id you previously received; the client will only deliver messages after it (reconnecting if necessary to reach it):

useSubscribeToStreams(
  { streamKey: `job.logs:${jobId}`, latestMessageId: lastSeenId },
  onMessage
);

Synced state graphs

When the backend stream is a StateStream, the server maintains a JSON object graph and emits granular mutations. On the client you get an observable that patches itself — no manual event handling.

useStreamStateObject (the easy path)

import { observer } from 'mobx-react-lite';
import { useStreamStateObject } from 'event-stream-react';

const Document = observer(({ id }: { id: string }) => {
  const doc = useStreamStateObject<{
    title?: string;
    comments?: { body: string }[];
  }>(`document:${id}`);

  return (
    <article>
      <h1>{doc.state.title ?? 'Untitled'}</h1>
      <ul>
        {(doc.state.comments ?? []).map((c, i) => (
          <li key={i}>{c.body}</li>
        ))}
      </ul>
    </article>
  );
});

• doc.state is the current graph. On connect it is populated from the server's snapshot (the prelude); thereafter it updates in place as mutations arrive.
• The optional generic types state for your editor.
• Wrap the component in observer (from mobx-react-lite) so it re-renders when state changes — state is a MobX observable. This is the standard way to read MobX state in a React component.

StreamStateObject (manual control)

useStreamStateObject is a thin convenience over mounting a StreamStateObject yourself. Do it manually when you need the instance to outlive a single component, or to inject it elsewhere. (See react-state-object for the mounting and injection APIs.)

import { StreamStateObject } from 'event-stream-react';

class DocumentStore extends StreamStateObject<{ title?: string }> {
  // add derived getters, actions, etc.
}

Low-level: EventStreamConnection

For advanced or non-hook usage you can open a connection directly. This is the primitive the manager builds on; most apps never need it.

import { EventStreamConnection } from 'event-stream-react';

const connection = new EventStreamConnection(
  ['job.progress:abc-123'],
  { url: '/streams', withCredentials: true }
);

const unsubscribe = connection.onMessage((envelope) => {
  console.log(envelope.message);
});

// later
unsubscribe();
connection.close();

Unlike the manager, a raw EventStreamConnection does not deduplicate across your app or coordinate reconnects with other subscribers — prefer the hooks unless you have a specific reason.

Utilities

The package also exports a few primitives used internally, in case they are useful:

• RedisStreamId — compare/min/max Redis stream ids (<ms>-<seq>).
• resolveJsonPathType, emptyContainerForPath — the client-side mirror of the backend's jsonPathTypes() container-type resolution.
• debounce — a TC39 (standard) method decorator.

Wire protocol

This package and event-stream/laravel share a wire contract. Keep the two on matching major versions. Each delivered message is the StreamEnvelope shown above; the resume cursor is a composite Last-Event-ID of the form streamKey1=redisId1;streamKey2=redisId2. State-stream mutations use the reserved message.type values state_stream_mutation_reset, state_stream_mutation_set, and state_stream_mutation_delete, which StreamStateObject handles for you.

Module formats & decorators

The package ships both ESM and CommonJS builds with TypeScript declarations, and works with any modern bundler (Vite, webpack, Next, etc.). The synced-graph layer uses TC39 standard decorators (via MobX and react-state-object); no experimentalDecorators setting is required in your app.

Publishing new versions

See PUBLISHING.md.

License

MIT