Перейти к содержимому
actor-ts
Русский

SSE (Server-Sent Events)

Это содержимое пока не доступно на вашем языке.

SseActor connects to and consumes a Server-Sent Events endpoint. It is a read-only client, not a server: it opens a long-lived HTTP GET, parses the text/event-stream wire format, and forwards each parsed event to a target actor. There are no commands and no outbound path — SSE is unidirectional, from server to client.

import { ActorSystem, Actor } from 'actor-ts';
import { SseActor, type SseEvent } from 'actor-ts/io';
import { SseOptions } from 'actor-ts/io';


// A collector receives each parsed event.
class Collector extends Actor<SseEvent> {
  override onReceive(ev: SseEvent): void {
    // ev.event — the `event:` field ('message' by default)
    // ev.data  — the `data:` payload (multiline joined with '\n')
    // ev.id    — the `id:` field, when the server sent one
    console.log(ev.event, ev.data, ev.id);
  }
}


const target = sys.spawnAnonymous(Collector);


const sseOptions = SseOptions.create()
  .withUrl('https://example.com/events')
  .withTarget(target)
  .withHeaders({ authorization: 'Bearer …' });
sys.spawnAnonymous(() => new SseActor(
  sseOptions,
));

The actor uses the global fetch (Bun, Node ≥ 18, and Deno all provide one), so there is no extra dependency to install.

Settings

Section titled “Settings”
interface SseOptionsType extends BrokerCommonOptionsType {
  url?:      string;                          // required — the SSE endpoint
  headers?:  Readonly<Record<string, string>>; // optional custom request headers
  target?:   ActorRef<SseEvent>;             // required — subscriber for inbound events
  idleTimeoutMs?:    number;                 // read-idle deadline; default 0 (off)
  connectTimeoutMs?: number;                 // connect deadline; default 0 (off)
}
SettingRequiredDescription
urlyesThe SSE endpoint the actor connects to (GET, Accept: text/event-stream).
targetyesThe actor that receives each parsed SseEvent.
headersnoExtra request headers — e.g. authorization for a protected feed.
idleTimeoutMsnoDeclare the stream lost after this long without a single inbound byte.
connectTimeoutMsnoAbandon a connect that has not produced response headers in time.

requiredOptions is ['url', 'target']; both must be present or the actor fails fast at startup.

Events the client parses

Section titled “Events the client parses”

Each parsed event is delivered as an SseEvent:

type SseEvent = {
  event: string;   // the `event:` field, or 'message' by default
  data:  string;   // the `data:` payload
  id?:   string;   // the `id:` field, when present
};

The client parses the SSE wire format — events separated by a blank line (\n\n), fields as field: value:

data: hello


event: tick
data: {"n":1}
id: 100


data: line-1
data: line-2

Parsing rules that follow the SSE spec:

  • Default event name. A block without an event: field arrives with event: 'message'.
  • Multiline data. Multiple data: lines in one block are joined with '\n' — the last block above yields data: 'line-1\nline-2'.
  • Comments ignored. Lines beginning with : (keepalive comments) are skipped.
  • id. Delivered as ev.id when the server sends one; otherwise undefined.

So the stream above delivers three events: { event: 'message', data: 'hello' }, one tick event with id '100', and the multiline { event: 'message', data: 'line-1\nline-2' }.

Reconnect and circuit breaker

Section titled “Reconnect and circuit breaker”

SseActor extends BrokerActor, so reconnect-with-backoff and the circuit breaker are inherited — configure them on the builder:

SseOptions.create()
  .withUrl('https://example.com/events')
  .withTarget(target)
  .withReconnect({ /* base delay, max delay, jitter … */ })
  .withCircuitBreaker(5, 30_000);   // failureThreshold, resetMs

When the stream ends or the connection drops observably, the base class’ reconnect machinery kicks in. Pass .withReconnect(false) to stop after the stream ends instead of retrying — handy for finite feeds or tests:

SseOptions.create()
  .withUrl(url)
  .withTarget(target)
  .withReconnect(false);   // stop when the stream closes

When the server just stops

Section titled “When the server just stops”

“Observably” is doing real work in that sentence. consume parks on reader.read(), and a server that vanishes without closing the response — a killed container, an expired NAT entry, a black-holing route — produces no done, no error and no bytes. SSE is read-only, so nothing else in the actor can tell that apart from a feed with nothing to report: the stream stays connected for as long as the process runs, and monitoring is blind.

idleTimeoutMs is the clock that ends it:

const sseOptions = SseOptions.create()
  .withUrl('https://example.com/events')
  .withTarget(target)
  .withIdleTimeoutMs(60_000)      // no bytes for 60 s → reconnect
  .withConnectTimeoutMs(10_000);  // no response headers in 10 s → fail the attempt

Any chunk refreshes it, including a comment — : ping\n\n parses to no event at all, which is precisely how a well-behaved server keeps an idle feed open. So set idleTimeoutMs comfortably above the server’s keepalive interval (15–30 s is the common convention); below it, the deadline severs healthy streams in a loop. Off by default, because only you know that interval.

connectTimeoutMs is the other half: fetch has no deadline of its own, so a server that accepts the TCP connection and never answers the request holds the actor in connecting indefinitely.

Both report the loss through the ordinary handleConnectionLost path, so backoff, the circuit breaker and BrokerDisconnected behave exactly as they do for a stream that ended — see BrokerActor base.

Configuration via HOCON

Section titled “Configuration via HOCON”

Settings resolve with the usual precedence — explicit options > HOCON > built-in defaults. The HOCON path is actor-ts.io.broker.sse:

actor-ts.io.broker.sse {
  url = "https://example.com/events"
  headers { authorization = "Bearer …" }
  idleTimeoutMs    = 60000
  connectTimeoutMs = 10000
}

target is an actor reference and so can only be set in code, not HOCON.

When to use SSE

Section titled “When to use SSE”

SseActor fits consuming an external SSE feed — anywhere a server pushes a one-way stream over plain HTTP and you want each event routed into the actor system:

  1. LLM token streams — consume a streaming completion endpoint and forward tokens to a collector.
  2. Market data / price ticks — a live feed pushed as events.
  3. CI / build logs — a long-running log stream tailed as it is produced.

Because it is just HTTP, it works through proxies, load balancers, and CDNs that don’t understand WebSocket.

Where to next

Section titled “Where to next”