Skip to content

AIEPhoenix/modern-fetch-stream

BranchesTags

Repository files navigation

modern-fetch-stream

npm version npm downloads bundle size license TypeScript

A lightweight Server-Sent Events client built on the Fetch API with automatic reconnection, last-event-id tracking, and explicit response / error classification.

SSE parsing is delegated to the spec-compliant eventsource-parser.

Why

The native EventSource API is limited: GET only, no custom headers, no request body, and very little control over retry strategy.

modern-fetch-stream keeps the good parts of fetch-based SSE clients while making retry semantics explicit:

  • Use fetch directly for POST requests, custom headers, and request bodies.
  • Classify responses separately from runtime errors with classifyResponse and classifyError.
  • Throw FatalError / RetriableError when you want the library to handle retry semantics for you.
  • Run anywhere fetch exists: browsers, Node.js 18+, Bun, and Deno.

Install

npm install modern-fetch-stream

1.0.0 breaking changes

1.0.0 is a breaking release. The main API changes are:

  • onerror is removed. Retry/fatal decisions now live in classifyResponse and classifyError.
  • Lifecycle callbacks are now onOpen, onMessage, and onClose.
  • onMessage can be async and is awaited serially.
  • The default response policy now accepts only 2xx text/event-stream responses.

See the changelog for the full release history, including the 1.0.1 state-machine and resource-handling fixes.

Quick start

import {
  FatalError,
  FetchEventSourceDecision,
  RetriableError,
  fetchEventSource,
} from 'modern-fetch-stream'

await fetchEventSource('/api/chat', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    Authorization: 'Bearer sk-...',
  },
  body: JSON.stringify({ prompt: 'Hello' }),

  classifyResponse(response) {
    if (response.ok) return FetchEventSourceDecision.Accept

    if (response.status >= 400 && response.status < 500 && response.status !== 429) {
      return FetchEventSourceDecision.Fatal
    }

    return FetchEventSourceDecision.Retry
  },

  onOpen(response) {
    console.log('stream opened', response.status)
  },

  onMessage(event) {
    console.log(event.data)
  },

  onClose(close) {
    console.log('stream closed', close.reason, close.receiveState)
  },

  classifyError(error) {
    if (error instanceof FatalError) return FetchEventSourceDecision.Fatal
    if (error instanceof RetriableError) return { retryAfter: error.retryAfter ?? 1000 }
    return FetchEventSourceDecision.Retry
  },
})

Migration from 0.x

If you used the pre-1.0.0 API, the main mapping is:

Before Now
onopen(response) to validate the response classifyResponse(response) to decide Accept / Retry / Fatal, then onOpen(response) for side effects
onmessage(event) onMessage(event)
onclose() onClose({ reason, receiveState })
onerror(error) returning a retry interval classifyError(error, receiveState) returning Retry, Fatal, or { retryAfter }

API

fetchEventSource(input, init): Promise<void>

Parameter Type Description
input RequestInfo | URL The resource to fetch.
init FetchEventSourceInit Fetch options extended with SSE callbacks and classifiers.

Reconnection caveat for Request inputs with a body. The library reuses the same input for every reconnection attempt. A Request whose body has been consumed by the first attempt cannot be replayed by native fetch and will throw on retry. For automatically reconnecting POST/PUT requests, prefer passing a URL string plus body (so the body string is sent fresh each time), or build a new Request per attempt outside the library.

At a high level, the library has three phases:

  1. classify the HTTP response
  2. consume SSE messages
  3. either close or retry
flowchart TD
    A["fetchEventSource(input, init)"] --> B["fetch(...)"]
    B --> C["classifyResponse(response)"]
    C -->|"Accept"| D["onOpen(response)"]
    C -->|"Retry / { retryAfter }"| E["discard response"]
    E --> F["schedule retry"]
    C -->|"Fatal"| G["reject with ResponseError"]
    D --> H["read SSE stream"]
    H --> I["onMessage(event)"]
    I --> H
    H -->|"EOF"| J["onClose({ reason: 'eof', receiveState })"]
    J --> K["resolve"]
    B -->|"network / runtime error"| L["classifyError(error, receiveState)"]
    D -->|"throw / reject"| L
    I -->|"throw / reject"| L
    J -->|"throw / reject"| L
    L -->|"Retry / { retryAfter }"| F
    L -->|"Fatal"| M["reject(error)"]
Loading

FetchEventSourceInit

Extends the standard RequestInit with the following:

Option Type Default Description
headers Record<string, string> {} Request headers. accept: text/event-stream is added automatically.
fetch typeof fetch globalThis.fetch Custom fetch implementation.
openWhenHidden boolean false Keep the connection alive when the page is hidden.
classifyResponse (response) => ResponseDecision Accept only 2xx text/event-stream responses Decides whether a newly received response should be accepted, retried, or treated as fatal.
onOpen (response) => void | Promise<void> — Called after classifyResponse accepts the response and before the body is consumed.
onMessage (event) => void | Promise<void> — Called for every SSE message, including custom event types. Async handlers are awaited serially.
onClose ({ reason, receiveState }) => void | Promise<void> — Called when the SSE request closes. reason is eof for stream completion and aborted for external cancellation. Throwing on eof routes through classifyError; throwing on aborted rejects directly.
classifyError (error, receiveState) => ErrorDecision Retry generic errors; fatal for ResponseError / FatalError; retry for RetriableError Decides whether an error should be retried or treated as fatal.

Execution order

On a successful stream, the callbacks run in this order:

classifyResponse -> onOpen -> onMessage... -> onClose({ reason: "eof" })

If any of these throw or reject, the error is routed through classifyError.

External abort follows this path:

AbortSignal -> onClose({ reason: "aborted" }) -> resolve

If onClose({ reason: "aborted" }) throws or rejects, the returned promise rejects directly instead of calling classifyError.

Successful completion timeline:

sequenceDiagram
    participant U as User
    participant L as modern-fetch-stream
    participant S as Server
    U->>L: fetchEventSource(...)
    L->>S: fetch request
    S-->>L: HTTP response
    L->>L: classifyResponse(response)
    L->>U: onOpen(response)
    loop for each SSE event
        S-->>L: SSE event
        L->>U: onMessage(event)
    end
    S-->>L: EOF
    L->>U: onClose({ reason: "eof", receiveState })
    L-->>U: Promise resolves
Loading

Abort timeline:

sequenceDiagram
    participant U as User
    participant A as AbortController
    participant L as modern-fetch-stream
    participant S as Server
    U->>L: fetchEventSource(..., { signal })
    L->>S: fetch request
    U->>A: abort()
    A-->>L: AbortSignal
    L->>U: onClose({ reason: "aborted", receiveState })
    L-->>U: Promise resolves
Loading

Error classes

The library exports four error classes, all inheriting from FetchEventSourceError:

import {
  FetchEventSourceError,
  ResponseError,
  FatalError,
  RetriableError,
} from 'modern-fetch-stream'
Class code Description
FetchEventSourceError — Abstract base class. Every subclass carries a unique string code for programmatic matching.
ResponseError RESPONSE_ERROR Wraps a rejected HTTP Response object, exposed via error.response.
FatalError FATAL_ERROR Signals an unrecoverable error — stops retrying immediately.
RetriableError RETRIABLE_ERROR Signals a transient error. Optional retryAfter (ms) overrides the current retry interval.

Other exports

import {
  EventStreamContentType,
  FetchEventSourceCloseReason,
  FetchEventSourceDecision,
  ReceiveState,
} from 'modern-fetch-stream'

// Type-level exports (useful for typing your own classifier functions):
import type {
  ErrorDecision,
  ResponseDecision,
  EventSourceMessage,
  FetchEventSourceClose,
  FetchEventSourceInit,
  FetchEventSourceDecisionValue,
  FetchEventSourceCloseReasonValue,
} from 'modern-fetch-stream'
  • EventStreamContentType: the standard text/event-stream MIME type.
  • FetchEventSourceCloseReason: runtime close-reason constants for Eof and Aborted.
  • FetchEventSourceDecision: runtime decision constants for Accept, Retry, and Fatal.
  • ReceiveState: final stream receive state passed to onClose and classifyError.
  • ErrorDecision / ResponseDecision: return types for classifyError and classifyResponse.
  • FetchEventSourceClose: payload type for the onClose callback.
  • EventSourceMessage: re-exported from eventsource-parser.

Decision constants

import { FetchEventSourceCloseReason, FetchEventSourceDecision } from 'modern-fetch-stream'

FetchEventSourceDecision.Accept // "accept"
FetchEventSourceDecision.Retry  // "retry"
FetchEventSourceDecision.Fatal  // "fatal"

FetchEventSourceCloseReason.Eof     // "eof"
FetchEventSourceCloseReason.Aborted // "aborted"
type ErrorDecision =
  | typeof FetchEventSourceDecision.Retry
  | typeof FetchEventSourceDecision.Fatal
  | { retryAfter: number }

type ResponseDecision =
  | typeof FetchEventSourceDecision.Accept
  | ErrorDecision

ReceiveState

ReceiveState.IDLE
ReceiveState.RECEIVED
ReceiveState.RECEIVED_NO_ID
  • IDLE: the stream closed before any message was received.
  • RECEIVED: at least one message with an id was received and last-event-id is available.
  • RECEIVED_NO_ID: messages were received, but there is no resumable last-event-id.

Response classification

Use classifyResponse when retry policy depends on the HTTP response itself:

await fetchEventSource('/api/stream', {
  classifyResponse(response) {
    if (response.ok) return FetchEventSourceDecision.Accept

    if (response.status === 429) {
      return { retryAfter: 5000 }
    }

    if (response.status >= 400 && response.status < 500) {
      return FetchEventSourceDecision.Fatal
    }

    return FetchEventSourceDecision.Retry
  },
})

If classifyResponse returns Retry, Fatal, or { retryAfter }, the current response is discarded immediately. Rejected responses become ResponseError instances when they terminate the stream.

flowchart TD
    A["response received"] --> B["classifyResponse(response)"]
    B -->|"Accept"| C["continue into onOpen + stream reading"]
    B -->|"Retry"| D["discard response"]
    D --> E["retry after current/server interval"]
    B -->|"{ retryAfter }"| F["discard response"]
    F --> G["retry after custom delay"]
    B -->|"Fatal"| H["discard response"]
    H --> I["reject with ResponseError"]
Loading

Error classification

Use classifyError when retry policy depends on runtime failures or exceptions thrown or rejected inside callbacks:

await fetchEventSource('/api/stream', {
  onClose(close) {
    if (close.reason === FetchEventSourceCloseReason.Eof) {
      throw new RetriableError('server closed early', 250)
    }
  },

  classifyError(error, receiveState) {
    if (receiveState === 'IDLE') {
      return { retryAfter: 2000 }
    }

    if (error instanceof FatalError) {
      return FetchEventSourceDecision.Fatal
    }

    return FetchEventSourceDecision.Retry
  },
})

If you omit classifyError, the defaults are:

  • ResponseError -> fatal
  • FatalError -> fatal
  • RetriableError -> retry, using retryAfter when provided
  • every other error -> retry

classifyError is only used for runtime failures. If classifyResponse directly returns Fatal or Retry, that decision is applied without calling classifyError.

External aborts are treated as normal shutdown: the returned promise resolves after onClose({ reason: "aborted" }). If that aborted close handler throws, the promise rejects directly.

flowchart TD
    A["runtime failure"] --> B["classifyError(error, receiveState)"]
    B -->|"Retry"| C["retry after current interval"]
    B -->|"{ retryAfter }"| D["retry after custom delay"]
    B -->|"Fatal"| E["reject(error)"]
Loading

Message handling and backpressure

onMessage is awaited serially. This means:

  • messages are processed in order
  • rejected async handlers flow into classifyError
  • slow handlers apply backpressure to the stream

If you want fire-and-forget work, do it explicitly:

onMessage(event) {
  void processLater(event)
}
flowchart LR
    A["event 1"] --> B["await onMessage(event 1)"]
    B --> C["event 2"]
    C --> D["await onMessage(event 2)"]
    D --> E["event 3"]
    E --> F["await onMessage(event 3)"]
Loading

Reconnection

On each retry the library automatically sends the last-event-id header with the id of the most recently received message, allowing the server to resume from where it left off.

The server can also control the retry interval by including a retry field in the event stream:

retry: 3000
data: hello

If your own classifyError or RetriableError does not specify retryAfter, the latest server-provided retry interval is used.

The library does not impose a built-in max retry count. If you want limits such as maxRetries, track that state outside the library and return FetchEventSourceDecision.Fatal when your policy is exhausted.

flowchart TD
    A["receive event with id"] --> B["store last-event-id"]
    B --> C["connection breaks or retry is requested"]
    C --> D["next fetch includes last-event-id header"]
    D --> E["server resumes from next event"]
Loading

Page visibility

In browsers, the connection is closed when the page becomes hidden and re-established when it becomes visible again. Set openWhenHidden: true to disable this behavior.

This feature is skipped automatically in non-browser environments.

Visibility-driven internal aborts do not surface as reason: "aborted". That close reason is reserved for caller-controlled cancellation via AbortSignal.

Request input support

fetchEventSource accepts a URL string, Request, or URL object as its first argument.

When you pass a Request:

  • Its headers are merged (lowercase-normalized) into the final header set.
  • Its signal is respected alongside any explicit signal in the options — either signal aborting will close the stream.
  • Explicit init options take precedence over the Request's properties.

The library normalizes all header names to lowercase so it can safely manage accept and last-event-id without collisions.

License

MIT

About

No description, website, or topics provided.

Resources

Readme

License

MIT license
Activity

Stars

1 star

Watchers

0 watching

Forks

0 forks
Report repository

Releases

6 tags

Packages

 
 
 

Contributors

Languages