Kotlin SDK - Stream Observer

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

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`

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. Example:

override fun onConnected() {
    logger.info("Nexus stream connected")
    metrics.increment("nexus.stream.connect")
}

Dispatch guarantees

Every callback is invoked through one guarded dispatcher inside the SDK. That gives three guarantees worth relying on:

A callback that throws cannot damage the SDK. The exception is absorbed, so stream control flow is unaffected. Without the guard, a throwing callback would cancel the stream coroutine and live updates would stop for the rest of the process’s life.
It is always reported. The failure is logged through NexusConfig.logger at warn, naming the callback: nexus: observer callback onConnected threw and was absorbed. Absorbing without reporting is the worse failure - the stream carries on while your instrumentation has silently stopped.
Cancellation still works. CancellationException is rethrown untouched, because it is how close() and disconnectStream() stop the stream.

Callbacks run on the SDK’s own coroutine scope, on Dispatchers.IO. Keep the work short: a blocking callback delays event processing on the stream path. Hand anything slow to your own scope. A test walks the SDK’s source and fails the build if any file outside the dispatcher touches a callback name, so the guard cannot be bypassed by a later change.

`onDisconnected`

fun onDisconnected(cause: Throwable?)

cause is the transport error when one occurred, and null for a clean server-side close - a deploy, or a WIF session rollover. Both reconnect; a connection that had stayed up for 30 seconds or more also resets the failure budget. 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. Example:

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

`onEvent`

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. Example:

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

`onReconnectAttempt`

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. Example:

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

`onFallback`

fun onFallback(reason: String)

When it fires: The stream gave up on immediate reconnection and TTL-based background sync is covering freshness in the meantime. The stream keeps trying: it waits out the next entry of the sseReconnectCooldown schedule (or the Retry-After delay for a rate limit) and reconnects. The reason value is one of the constants below.

Reason constant	Value	Meaning
`FALLBACK_REASON_MAX_ERRORS`	`"max-errors"`	3 consecutive non-quarantine transport failures
`FALLBACK_REASON_RATE_LIMITED`	`"rate-limited"`	A plain HTTP 429 (without a quarantine body) on the stream path

Example:

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`

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. Important: A quarantine is a cooperative pause, not a failure. The SDK’s internal failure counter is not incremented when a quarantine occurs. This means a quarantine period between two successful connections does not consume any of the 3-failure budget used by onFallback. Parameters:

Parameter	Type	Description
`reason`	`String`	Human-readable reason supplied by the server
`expiresAt`	`Instant`	Absolute time when the quarantine lifts

Example:

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`

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 (both TTL background and manual). Getters continue to return the last successfully cached values. Example:

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

Full observer example

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())
    }
}

Overview

Backend SDKs

Browser SDKs

OpenFeature Providers

Kotlin SDK - Stream Observer

Callback reference

`onConnected`

Dispatch guarantees

`onDisconnected`

`onEvent`

`onReconnectAttempt`

`onFallback`

`onQuarantined`

`onBillingOverdue`

Full observer example

​Callback reference

​onConnected

​Dispatch guarantees

​onDisconnected

​onEvent

​onReconnectAttempt

​onFallback

​onQuarantined

​onBillingOverdue

​Full observer example

Callback reference

`onConnected`

Dispatch guarantees

`onDisconnected`

`onEvent`

`onReconnectAttempt`

`onFallback`

`onQuarantined`

`onBillingOverdue`

Full observer example