> ## Documentation Index
> Fetch the complete documentation index at: https://docs.westyx.cloud/llms.txt
> Use this file to discover all available pages before exploring further.

# Spring Boot SDK - Stream Observer

> NexusStreamObserver reference and Spring bean wiring for the Spring Boot SDK.

Pass a `NexusStreamObserver` via `NexusConfig.Builder#observer(...)` 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 `NexusLogger` output by substring match.

The observer is **opt-in**. When `NexusConfig.observer()` is `null` the SDK uses `NexusStreamObserver.NOOP` at zero overhead. The auto-configuration also picks up any `NexusStreamObserver` bean automatically.

## The interface

```java theme={null}
public interface NexusStreamObserver {
    String FALLBACK_REASON_MAX_ERRORS   = "max-errors";
    String FALLBACK_REASON_RATE_LIMITED = "rate-limited";

    NexusStreamObserver NOOP = new NexusStreamObserver() {};

    default void onConnected() {}
    default void onDisconnected(Throwable cause) {}
    default void onEvent(String name) {}
    default void onReconnectAttempt(int attempt) {}
    default void onFallback(String reason) {}
    default void onQuarantined(String reason, java.time.Instant expiresAt) {} // v0.4.0
}
```

Every method has an empty `default` - implement only the hooks you care about.

## Hook semantics

| Hook                                          | Fires when                                                                                                                                                         | Argument                                                       |
| --------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------------------------------------------------------------- |
| `onConnected()`                               | `GET /v1/stream` returns `200 OK` and the SDK is ready to read events                                                                                              | none                                                           |
| `onDisconnected(cause)`                       | An active stream connection ends                                                                                                                                   | `null` for clean shutdown, the underlying exception otherwise  |
| `onEvent(name)`                               | A whitelisted SSE event arrives, **before** the SDK runs the event-triggered sync                                                                                  | event name string (e.g. `"config.updated"`)                    |
| `onReconnectAttempt(n)`                       | The SDK is about to back off and retry after a transport error                                                                                                     | 1-based attempt counter (max 3 with current threshold)         |
| `onFallback(reason)`                          | The SDK either gives up and switches to TTL polling, **or** is about to sleep through a server-indicated `Retry-After` rate-limit window before reconnecting       | `FALLBACK_REASON_MAX_ERRORS` or `FALLBACK_REASON_RATE_LIMITED` |
| `onQuarantined(reason, expiresAt)` *(v0.4.0)* | Server returned HTTP 429 with a quarantine body. Fired for both sync and stream 429s. Background sync pauses until `expiresAt`; the stream sleeps then reconnects. | server-supplied reason string + expiry `Instant`               |

### `FALLBACK_REASON_RATE_LIMITED` semantics (v0.3.1+)

Starting in v0.3.1 the `"rate-limited"` reason fires in **two** distinct situations:

1. **Terminal fallback** - the 429 response had **no** parseable `Retry-After` header. The SDK terminates the stream loop after firing `onFallback("rate-limited")` and TTL polling takes over.
2. **Per-episode notification before reconnect** - the 429 response carried a parseable `Retry-After` header. The SDK fires `onFallback("rate-limited")`, sleeps the clamped delay (`[5 s, 5 min]`, interrupt-aware) and then **automatically reconnects**. The SSE thread does NOT exit, and the rate-limit 429 does NOT count toward `MAX_STREAM_ERRORS`.

Treat the reason as a **rate-limit signal**, not as a permanent state change. Use `client.isStreamConnected()` (or `onConnected` / `onDisconnected`) to track the actual connection state.

`FALLBACK_REASON_MAX_ERRORS` retains its prior meaning: a terminal switch to TTL polling after 3 consecutive transport errors, and the SSE thread exits.

## Threading contract

<Warning>
  Observer methods are called **synchronously from the SDK's stream-reader Thread**. They MUST return promptly. The SDK does not protect against blocking observers.
</Warning>

If your observer does anything substantial (network I/O, lock contention, disk writes), copy the argument and dispatch the work to your own executor before returning.

## Spring Boot bean wiring

The auto-configuration picks up any `NexusStreamObserver` bean automatically:

```java theme={null}
@Configuration
class NexusObservability {

    @Bean
    NexusStreamObserver nexusObserver(MeterRegistry meters) {
        return new NexusStreamObserver() {
            @Override public void onConnected() {
                meters.counter("nexus.stream.connected").increment();
            }
            @Override public void onDisconnected(Throwable cause) {
                meters.counter("nexus.stream.disconnected").increment();
            }
            @Override public void onEvent(String name) {
                meters.counter("nexus.stream.event", "name", name).increment();
            }
            @Override public void onReconnectAttempt(int attempt) {
                meters.counter("nexus.stream.reconnect").increment();
            }
            @Override public void onFallback(String reason) {
                meters.counter("nexus.stream.fallback", "reason", reason).increment();
            }
        };
    }
}
```

No additional `NexusConfig` wiring is needed - the auto-config calls `builder.observer(observer)` for you.

## Counter pattern

```java theme={null}
public class CounterObserver implements NexusStreamObserver {
    private final AtomicLong events = new AtomicLong();
    private final AtomicLong reconnects = new AtomicLong();
    private final AtomicLong fallbacks = new AtomicLong();

    public long events() { return events.get(); }
    public long reconnects() { return reconnects.get(); }
    public long fallbacks() { return fallbacks.get(); }

    @Override public void onEvent(String name)            { events.incrementAndGet(); }
    @Override public void onReconnectAttempt(int attempt) { reconnects.incrementAndGet(); }
    @Override public void onFallback(String reason)       { fallbacks.incrementAndGet(); }
}
```