> ## 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.

# Kotlin SDK - Stream Observer

> NexusObserver callback reference for the Kotlin SDK.

`NexusObserver` is an interface with default no-op implementations for every method. You only need to override the callbacks you care about.

```kotlin theme={null}
interface NexusObserver {
    fun onConnected()
    fun onDisconnected(cause: Throwable?)
    fun onEvent(name: String)
    fun onReconnectAttempt(attempt: Int)
    fun onFallback(reason: String)
    fun onQuarantined(reason: String, expiresAt: Instant)
    fun onBillingOverdue()
}
```

Pass your implementation via `NexusConfig.observer`. The default is `NoopNexusObserver`, which discards all events.

## Callback reference

### `onConnected`

```kotlin theme={null}
fun onConnected()
```

**When it fires:** The SSE connection to `/v1/stream` was established successfully and the SDK received HTTP 200. `streamStatus` transitions to `CONNECTED` before this callback is invoked.

```kotlin theme={null}
override fun onConnected() {
    logger.info("Nexus stream connected")
    metrics.increment("nexus.stream.connect")
}
```

### `onDisconnected`

```kotlin theme={null}
fun onDisconnected(cause: Throwable?)
```

**When it fires:** The SSE connection was lost. `cause` is the exception that triggered the disconnect, or `null` if the disconnect was clean (server closed the stream or `disconnectStream()` was called). Also fires on a plain 429 rate-limit before the retry delay.

```kotlin theme={null}
override fun onDisconnected(cause: Throwable?) {
    if (cause != null) {
        logger.warn("Nexus stream disconnected: ${cause.message}")
    } else {
        logger.debug("Nexus stream closed cleanly")
    }
}
```

### `onEvent`

```kotlin theme={null}
fun onEvent(name: String)
```

**When it fires:** The SSE stream received a named event from the server (for example `config_updated`, `flag_toggled`). The SDK has already started a background `sync()` call by the time this callback fires. Note: `auth_expiring` is handled internally and never surfaced here.

```kotlin theme={null}
override fun onEvent(name: String) {
    logger.debug("Nexus event received: $name")
    metrics.increment("nexus.stream.event", tag("event", name))
}
```

### `onReconnectAttempt`

```kotlin theme={null}
fun onReconnectAttempt(attempt: Int)
```

**When it fires:** A transient failure occurred and the SDK is about to attempt reconnection. `attempt` is 1-indexed (first reconnect = 1). Called after the back-off delay has elapsed, immediately before the new connection attempt.

### `onFallback`

```kotlin theme={null}
fun onFallback(reason: String)
```

**When it fires:** The stream loop has given up after 3 consecutive failures and will no longer attempt to reconnect. TTL-based background sync continues to run.

| Reason constant                | Value            | Meaning                                              |
| ------------------------------ | ---------------- | ---------------------------------------------------- |
| `FALLBACK_REASON_MAX_ERRORS`   | `"max-errors"`   | 3 consecutive non-quarantine failures                |
| `FALLBACK_REASON_RATE_LIMITED` | `"rate-limited"` | (reserved, not currently used as a fallback trigger) |

```kotlin theme={null}
override fun onFallback(reason: String) {
    logger.error("Nexus stream in fallback mode (reason=$reason) - relying on TTL sync")
    alerting.page("nexus-stream-fallback", reason)
}
```

### `onQuarantined`

```kotlin theme={null}
fun onQuarantined(reason: String, expiresAt: Instant)
```

**When it fires:** The server returned HTTP 429 with a quarantine payload (`{"error":"quarantined",...}`). This happens on both the sync path and the stream path. The SDK automatically pauses until `expiresAt` before retrying.

<Note>
  A quarantine is a cooperative pause, not a failure. The SDK's internal failure counter is **not** incremented when a quarantine occurs.
</Note>

```kotlin theme={null}
override fun onQuarantined(reason: String, expiresAt: Instant) {
    val seconds = expiresAt.epochSecond - Instant.now().epochSecond
    logger.warn("Nexus quarantined for ${seconds}s: $reason")
    metrics.increment("nexus.quarantine")
}
```

### `onBillingOverdue`

```kotlin theme={null}
fun onBillingOverdue()
```

**When it fires:** The sync endpoint returned HTTP 402. The tenant has overdue invoices and the server is refusing to serve fresh data. The SDK sets an internal billing-overdue flag that suppresses further sync attempts. Getters continue to return the last successfully cached values.

```kotlin theme={null}
override fun onBillingOverdue() {
    logger.error("Nexus: tenant billing overdue - contact support")
    alerting.page("nexus-billing-overdue")
}
```

## Full observer example

```kotlin theme={null}
class ProductionNexusObserver(
    private val logger: Logger,
    private val metrics: MetricsClient,
    private val alerting: AlertingClient,
) : NexusObserver {

    override fun onConnected() {
        logger.info("Nexus stream connected")
        metrics.gauge("nexus.stream.up", 1)
    }

    override fun onDisconnected(cause: Throwable?) {
        logger.warn("Nexus stream disconnected", cause)
        metrics.gauge("nexus.stream.up", 0)
    }

    override fun onEvent(name: String) {
        logger.debug("Nexus event: $name")
        metrics.increment("nexus.event", tag("name", name))
    }

    override fun onReconnectAttempt(attempt: Int) {
        logger.info("Nexus stream reconnect #$attempt")
    }

    override fun onFallback(reason: String) {
        logger.error("Nexus stream fallback: $reason")
        alerting.page("nexus-stream-down", mapOf("reason" to reason))
    }

    override fun onQuarantined(reason: String, expiresAt: Instant) {
        logger.warn("Nexus quarantined until $expiresAt: $reason")
        metrics.increment("nexus.quarantine")
    }

    override fun onBillingOverdue() {
        logger.error("Nexus billing overdue")
        alerting.page("nexus-billing-overdue", emptyMap())
    }
}
```