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.

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

`onDisconnected`

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.

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.

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.

`onFallback`

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)

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.

A quarantine is a cooperative pause, not a failure. The SDK’s internal failure counter is not incremented when a quarantine occurs.

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. Getters continue to return the last successfully cached values.

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`

`onDisconnected`

`onEvent`

`onReconnectAttempt`

`onFallback`

`onQuarantined`

`onBillingOverdue`

Full observer example

​Callback reference

​onConnected

​onDisconnected

​onEvent

​onReconnectAttempt

​onFallback

​onQuarantined

​onBillingOverdue

​Full observer example

Callback reference

`onConnected`

`onDisconnected`

`onEvent`

`onReconnectAttempt`

`onFallback`

`onQuarantined`

`onBillingOverdue`

Full observer example