Angular SDK - Stream Observer

Pass a NexusStreamObserver via Config.streamObserver to receive structured callbacks for every SSE lifecycle transition. The observer is the canonical way to wire SDK state into ops dashboards, custom telemetry, or load-test tooling - far more reliable than scraping Logger output by substring match. The observer is opt-in. When Config.streamObserver is undefined the SDK uses NOOP_STREAM_OBSERVER at zero overhead.

The interface

export interface NexusStreamObserver {
  readonly onConnected?: () => void;
  readonly onDisconnected?: (cause: unknown | null) => void;
  readonly onEvent?: (name: string) => void;
  readonly onReconnectAttempt?: (attempt: number) => void;
  readonly onFallback?: (reason: string) => void;
  /** v0.4.0+ - fires when the server returns 429 quarantined on sync or stream pre-flight. */
  readonly onQuarantined?: (reason: string, expiresAt: Date) => void;
}

Every method is optional - the SDK calls only the ones you set. You can write { onEvent: (n) => count(n) } and ignore the rest.

Hook semantics

Hook	Fires when	Argument
`onConnected()`	The fetch stream connects and the server confirms the response body is readable (initial connect or an SDK reconnect succeeds)	none
`onDisconnected(cause)`	The connection ends - `disconnectStream()`, transport error, or 3-strike fallback	`null` for clean disconnect, the underlying `Error` for transport failures
`onEvent(name)`	A whitelisted SSE event arrives, before the SDK runs the event-triggered sync	event name string (e.g. `"config.updated"`)
`onReconnectAttempt(n)`	A transient stream transport error fires (the SDK will auto-reconnect with backoff)	1-based counter, max 3
`onFallback(reason)`	The SDK gives up after 3 consecutive errors, hits a non-quarantine 429 on stream pre-flight, or schedules a 5xx sync retry	`FALLBACK_REASON_MAX_ERRORS`, `FALLBACK_REASON_RATE_LIMITED`, or `FALLBACK_REASON_TRANSPORT`
`onQuarantined(reason, expiresAt)` (v0.4.0)	The server returns HTTP 429 with `{"error":"quarantined"}` on sync or stream pre-flight	`reason: string` (server-supplied quarantine kind), `expiresAt: Date` (when the SDK will reconnect)

`onFallback` reasons

import {
  FALLBACK_REASON_MAX_ERRORS,
  FALLBACK_REASON_RATE_LIMITED,
  FALLBACK_REASON_TRANSPORT, // v0.3.1+
} from '@westyx-nexus/sdk-angular';

Constant	When
`'max-errors'`	3 consecutive stream transport errors - backoff exhausted
`'rate-limited'`	The SDK’s `fetch` pre-flight to the stream URL returned `429 Too Many Requests`. The SDK parses `Retry-After` and schedules a reconnect; the observer fires once per 429 episode. (v0.3.1+)
`'transport'`	A 5xx response from `/sync` triggered the exponential-backoff retry loop. The observer fires once per retry-scheduling; reset when the next sync succeeds. (v0.3.1+)

Angular zone integration

Observer methods are called inside NgZone.run(...) automatically - the same wrapping the SDK already uses for its stream event handlers. This means:

Signals updated inside an observer method trigger change detection
Component reads inside an observer method see the latest state
You do NOT need to wrap the observer body manually

Threading contract

Observer methods MUST return promptly. The SDK calls them synchronously from the stream reader loop (after the NgZone.run(...) wrap). If your observer does anything substantial (network I/O, structured logging fan-out), copy the argument and dispatch the work to a future microtask.

Counter pattern

import { provideNexus, NexusStreamObserver } from '@westyx-nexus/sdk-angular';

const counters = {
  events: 0,
  reconnects: 0,
  fallbacks: 0,
};

const observer: NexusStreamObserver = {
  onEvent:            ()  => counters.events++,
  onReconnectAttempt: ()  => counters.reconnects++,
  onFallback:         (r) => { counters.fallbacks++; console.warn('SSE fallback:', r); },
};

bootstrapApplication(AppComponent, {
  providers: [
    provideNexus({
      baseUrl: '...',
      apiKey:  'wxp_...',
      streamObserver: observer,
    }),
  ],
});

Active-connection signal pattern

Maintain a “live” indicator backed by an Angular signal:

import { signal } from '@angular/core';
import { NexusStreamObserver } from '@westyx-nexus/sdk-angular';

export const streamConnected = signal(false);

export const observer: NexusStreamObserver = {
  onConnected:    () => streamConnected.set(true),
  onDisconnected: () => streamConnected.set(false),
  onFallback:     () => streamConnected.set(false),
};

Then in a template:

<span class="status" [class.active]="streamConnected()">
  {{ streamConnected() ? 'Live updates on' : 'Polling fallback' }}
</span>

Overview

Backend SDKs

Browser SDKs

OpenFeature Providers

Angular SDK - Stream Observer

The interface

Hook semantics

`onFallback` reasons

Angular zone integration

Threading contract

Counter pattern

Active-connection signal pattern

​The interface

​Hook semantics

​onFallback reasons

​Angular zone integration

​Threading contract

​Counter pattern

​Active-connection signal pattern

The interface

Hook semantics

`onFallback` reasons

Angular zone integration

Threading contract

Counter pattern

Active-connection signal pattern