git.iko.soy/opencode-android
on 66d239c help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
181 lines 9.0 kB history raw
1package soy.iko.opencode.data.network
3import soy.iko.opencode.data.model.BusEvent
4import android.util.Log
5import io.ktor.client.HttpClient
6import io.ktor.client.plugins.timeout
7import io.ktor.client.plugins.sse.sse
8import io.ktor.sse.ServerSentEvent
9import kotlinx.coroutines.CancellationException
10import kotlinx.coroutines.CoroutineScope
11import kotlinx.coroutines.channels.BufferOverflow
12import kotlinx.coroutines.channels.Channel
13import kotlinx.coroutines.channels.ChannelResult
14import kotlinx.coroutines.flow.Flow
15import kotlinx.coroutines.flow.MutableStateFlow
16import kotlinx.coroutines.flow.SharedFlow
17import kotlinx.coroutines.flow.StateFlow
18import kotlinx.coroutines.flow.SharingStarted
19import kotlinx.coroutines.flow.asStateFlow
20import kotlinx.coroutines.flow.buffer
21import kotlinx.coroutines.flow.channelFlow
22import kotlinx.coroutines.flow.produceIn
23import kotlinx.coroutines.flow.shareIn
24import kotlinx.coroutines.isActive
25import kotlinx.coroutines.selects.onTimeout
26import kotlinx.coroutines.selects.select
27import java.io.IOException
28import kotlin.random.Random
30/**
31 * Owns the single long-lived `GET /event` SSE subscription and exposes it as a hot
32 * [SharedFlow] of [BusEvent]. The connection is opened when the first collector
33 * subscribes (WhileSubscribed) and reconnects with backoff on drop.
34 *
35 * Ordering contract: callers that send a prompt must already be collecting [events]
36 * (directly or via the repository) so early `message.part.updated` deltas aren't missed.
37 */
38@OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class)
39class EventStreamClient(
40 private val client: HttpClient,
41 scope: CoroutineScope,
42) {
43 enum class ConnectionState { Disconnected, Connecting, Connected }
45 private val _state = MutableStateFlow(ConnectionState.Disconnected)
46 val state: StateFlow<ConnectionState> = _state.asStateFlow()
48 val events: SharedFlow<BusEvent> = stream()
49 // Buffer so a slow subscriber (e.g. a reducer under lock) can't suspend send()
50 // and stall the SSE read loop. Drop oldest on overflow — the next event carries
51 // the freshest state, so a dropped interim delta is harmless.
52 .buffer(NetworkConfig.sseEventBufferCapacity, BufferOverflow.DROP_OLDEST)
53 .shareIn(
54 scope = scope,
55 started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
56 replay = 0,
57 )
59 /** Conflated channel used to cut the reconnect backoff short when network returns. */
60 private val reconnectSignal = Channel<Unit>(Channel.CONFLATED)
62 /** Returns true if the exception is an SSE auth failure (401/403) that should stop retrying. */
63 private fun isSseAuthFailure(e: Throwable): Boolean =
64 e is io.ktor.client.plugins.ClientRequestException &&
65 (e.response.status.value == 401 || e.response.status.value == 403)
67 /** Request an immediate reconnect, skipping any in-progress backoff. */
68 fun triggerReconnect() { reconnectSignal.trySend(Unit) }
70 /**
71 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
72 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
73 * clean stream close or scope cancellation.
74 */
75 private suspend fun readSseEvents(
76 scope: CoroutineScope,
77 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
78 send: suspend (BusEvent) -> Unit,
79 ) {
80 while (scope.isActive) {
81 // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
82 // read), so a socket the peer never closed would otherwise hang until OS TCP
83 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
84 // within the idle timeout, treat the connection as half-open and drop it.
85 val result: ChannelResult<ServerSentEvent>? = select {
86 events.onReceiveCatching { it }
87 onTimeout(NetworkConfig.sseIdleTimeoutMs) { null }
88 }
89 if (result == null) throw IOException("SSE idle timeout, reconnecting")
90 val sse = result.getOrNull() ?: break
91 val data = sse.data
92 if (data == null) {
93 // A comment or keep-alive event (no data field) resets the idle watchdog
94 // — servers send `: ping` comments to keep the connection alive through
95 // proxies. Without this, a server that only sends comments would still
96 // hit the idle timeout and reconnect unnecessarily.
97 continue
98 }
99 val event = try {
100 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
101 } catch (e: kotlinx.serialization.SerializationException) {
102 Log.d("EventStream", "Skipping unparseable SSE event", e)
103 continue
104 }
105 if (_state.value != ConnectionState.Connected) {
106 _state.value = ConnectionState.Connected
107 }
108 send(event)
109 }
110 }
112 private fun stream(): Flow<BusEvent> = channelFlow {
113 val scope = this
114 // Drain any stale reconnect signal from a previous collection cycle so it
115 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
116 // stopped the upstream while triggerReconnect() queued a signal).
117 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
118 var backoffMs = NetworkConfig.sseInitialBackoffMs
119 while (isActive) {
120 _state.value = ConnectionState.Connecting
121 try {
122 // The SSE stream is long-lived: disable the request-level and socket
123 // timeouts (the client default is 60s for REST) in the request config
124 // so the connection isn't killed mid-stream. The idle-timeout watchdog
125 // below handles half-open connections instead.
126 client.sse(
127 "event",
128 request = {
129 timeout {
130 requestTimeoutMillis = Long.MAX_VALUE
131 // Disable the socket (read) timeout entirely for the SSE stream.
132 // A finite value would kill the connection whenever the server
133 // is quiet for longer than that (idle sessions, slow agent runs),
134 // causing a reconnect storm. Half-open detection is handled by
135 // the idle-timeout watchdog below plus OkHttp's pingInterval on
136 // HTTP/2 connections.
137 socketTimeoutMillis = Long.MAX_VALUE
138 }
139 },
140 ) {
141 backoffMs = NetworkConfig.sseInitialBackoffMs
142 // Clear any stale reconnect signal from a triggerReconnect() call
143 // made while the previous stream was healthy, so it doesn't suppress
144 // the next legitimate backoff.
145 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
146 // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
147 // element can be raced against an idle timeout via select.
148 val events = incoming.produceIn(scope)
149 try {
150 readSseEvents(scope, events) { send(it) }
151 } finally {
152 events.cancel()
153 }
154 }
155 } catch (c: CancellationException) {
156 throw c
157 } catch (e: Exception) {
158 // If the scope was cancelled (e.g. connection close), the closed-client
159 // exception is expected — don't log a spurious "stream error" warning.
160 if (isSseAuthFailure(e)) {
161 // Credentials are wrong — retrying won't help. Log and stop.
162 Log.w("EventStream", "SSE auth failed (401/403), stopping retries", e)
163 _state.value = ConnectionState.Disconnected
164 break
165 } else if (isActive) {
166 Log.w("EventStream", "SSE stream error, will retry", e)
167 }
168 }
169 _state.value = ConnectionState.Disconnected
170 if (!isActive) break
171 // Wait for the backoff, but allow a reconnect signal to cut it short.
172 var signaled = false
173 val jitter = ((backoffMs * 0.2) * (Random.nextDouble() * 2 - 1)).toLong()
174 select {
175 reconnectSignal.onReceive { signaled = true }
176 onTimeout(backoffMs + jitter) { /* normal backoff elapsed */ }
177 }
178 backoffMs = if (signaled) NetworkConfig.sseInitialBackoffMs else (backoffMs * 2).coerceAtMost(NetworkConfig.sseMaxBackoffMs)
179 }
180 }