Kotlin SDK - SSE Live Updates

The SDK can maintain a persistent Server-Sent Events (SSE) connection to GET /v1/stream. When the Nexus platform publishes a change (config edit, secret rotation, flag toggle), the server sends an event and the SDK refreshes its local snapshot.

Starting the stream

val client = NexusClient.create(config)
client.connectStream()

connectStream() is non-blocking: it launches a coroutine on the SDK’s internal Dispatchers.IO scope and returns immediately. It is also idempotent - calling it while a stream is running leaves that stream in place and logs at debug. Starting a second one would leave two streams live with the first no longer reachable, so nothing short of close() could stop it. On a closed client it raises NexusClosedException rather than quietly doing nothing.

Observing stream state

launch {
    client.streamStatus.collect { status ->
        println("Stream: $status")
    }
}

streamStatus is a StateFlow<StreamStatus>:

Value	Meaning
`CONNECTED`	SSE connection established, receiving events
`FALLBACK`	The stream gave up for now and TTL polling is covering freshness; a reconnect is scheduled
`DISCONNECTED`	Not connected: never started, between reconnect attempts, or stopped

Deadlines

Deadline	Value	Why
connect	10 s	Establishing the connection must not hang.
socket idle	90 s	The longest gap between two reads before the connection is treated as dead.
request	none	An SSE response is open-ended; a request deadline would tear down a healthy connection on a fixed schedule.

The idle deadline is what distinguishes “quiet” from “gone”. The backend writes an SSE : ping comment every 30 seconds, and the read loop consumes those comments, so any live connection resets the timer three times over. A connection silently dropped by a load balancer produces no bytes at all and is detected within 90 seconds instead of reading as healthy indefinitely. Lifting the read deadline does not lift the connect deadline - they are separate settings.

Reconnecting

After a clean server close

The backend ends the stream at WIF session expiry and on every deploy, so a clean end-of-stream is the normal case rather than an edge case. The read loop treats it as a disconnect - firing observer.onDisconnected(null), with a null cause because there was no error - and reconnects. A connection that stayed up for 30 seconds or more counts as healthy: the failure budget and the polling schedule both restart. A deploy therefore costs one quick reconnect rather than pushing the client towards TTL polling.

After a transport error

Back-off starts at 1 second and doubles: 1 s, 2 s, 4 s, …, capped at 30 seconds.
observer.onDisconnected(cause) fires with the error.
observer.onReconnectAttempt(attempt) fires before each attempt (1-indexed).
After 3 consecutive failures the SDK calls observer.onFallback("max-errors"), sets streamStatus to FALLBACK, and waits out the next entry of the sseReconnectCooldown schedule before trying again. TTL-based background sync covers freshness meanwhile.

failure 1 -> onDisconnected -> wait 1s -> reconnect (attempt 1)
failure 2 -> onDisconnected -> wait 2s -> reconnect (attempt 2)
failure 3 -> onDisconnected -> onFallback("max-errors") -> status FALLBACK
          -> wait sseReconnectCooldown[0] (default 5 min) -> reconnect
          -> still failing -> wait [1] (10 min) -> ... -> stays at the last entry

The stream loop never exits on its own. It runs until disconnectStream() or close() cancels it. An over-long SSE line (above 64 KiB) or a line truncated by a mid-message close is a transport error and reconnects like any other.

Quarantine (HTTP 429 with a quarantine body)

A quarantine is a server-driven cooperative pause. Unlike a transport error, it does not consume the failure budget.

The expires_at deadline is clamped to at most 24 hours from now.
observer.onQuarantined(reason, expiresAt) is called with the clamped value, and streamStatus becomes FALLBACK.
The stream waits until the deadline (minimum 5 seconds), then reconnects as if no failure had occurred.

connected -> quarantine 429 -> onQuarantined -> wait until expiresAt
                                             -> reconnect (failure counter unchanged)
                                             -> connected again

Rate limit (HTTP 429 without a quarantine body)

A plain 429 means the server is temporarily rate-limiting this client.

The Retry-After header (in seconds) is used as the delay, with a 5-second floor.
observer.onFallback("rate-limited") and observer.onDisconnected(null) are called, and streamStatus becomes FALLBACK.
The stream reconnects after the delay. This does not consume the failure budget either.

`auth_expiring` event for WIF sessions

When WIF is active, the server sends an auth_expiring event before the session token expires. The SDK intercepts it and refreshes the session in the background - it does not trigger a sync and does not call onEvent, so your observer sees no callback for it. This gives zero-downtime token rotation for long-running WIF sessions.

Observer callbacks are guarded

Every callback is invoked through a single guarded dispatcher, so a callback that throws is absorbed, reported through the SDK logger at warn, and cannot stop the stream. CancellationException is rethrown untouched, because it is how close() stops the stream. See Stream observer.

Disconnecting the stream

client.disconnectStream()

Cancels the stream and sets streamStatus to DISCONNECTED. The HTTP client and TTL sync are unaffected, and connectStream() starts a fresh stream afterwards.

Shutting down completely

client.close()

Cancels the coroutine scope (including the stream), closes both HTTP clients and the WIF credential source, and removes materialised file secrets. close() is idempotent and final - connectStream() afterwards raises NexusClosedException. There is no need to call disconnectStream() first. NexusClient is AutoCloseable, so use { } works:

NexusClient.create(config).use { client ->
    client.connectStream()
    // ...
}

Example: reacting to status changes

import dev.westyx.nexus.StreamStatus
import kotlinx.coroutines.launch

launch {
    client.streamStatus.collect { status ->
        when (status) {
            StreamStatus.CONNECTED    -> metrics.gauge("nexus.stream", 1)
            StreamStatus.DISCONNECTED -> metrics.gauge("nexus.stream", 0)
            StreamStatus.FALLBACK     -> {
                metrics.gauge("nexus.stream", 0)
                alerting.send("Nexus stream in fallback - relying on TTL sync until it reconnects")
            }
        }
    }
}

client.connectStream()

Overview

Backend SDKs

Browser SDKs

OpenFeature Providers

Kotlin SDK - SSE Live Updates

Starting the stream

Observing stream state

Deadlines

Reconnecting

After a clean server close

After a transport error

Quarantine (HTTP 429 with a quarantine body)

Rate limit (HTTP 429 without a quarantine body)

`auth_expiring` event for WIF sessions

Observer callbacks are guarded

Disconnecting the stream

Shutting down completely

Example: reacting to status changes

​Starting the stream

​Observing stream state

​Deadlines

​Reconnecting

​After a clean server close

​After a transport error

​Quarantine (HTTP 429 with a quarantine body)

​Rate limit (HTTP 429 without a quarantine body)

​auth_expiring event for WIF sessions

​Observer callbacks are guarded

​Disconnecting the stream

​Shutting down completely

​Example: reacting to status changes

Starting the stream

Observing stream state

Deadlines

Reconnecting

After a clean server close

After a transport error

Quarantine (HTTP 429 with a quarantine body)

Rate limit (HTTP 429 without a quarantine body)

`auth_expiring` event for WIF sessions

Observer callbacks are guarded

Disconnecting the stream

Shutting down completely

Example: reacting to status changes