git.iko.soy/opencode-android
on d643c13 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
200 lines 10 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 private val idleTimeoutMs: Long = NetworkConfig.sseIdleTimeoutMs,
43 private val initialBackoffMs: Long = NetworkConfig.sseInitialBackoffMs,
44 private val maxBackoffMs: Long = NetworkConfig.sseMaxBackoffMs,
45 private val bufferCapacity: Int = NetworkConfig.sseEventBufferCapacity,
46) {
47 enum class ConnectionState { Disconnected, Connecting, Connected, Failed }
49 private val _state = MutableStateFlow(ConnectionState.Disconnected)
50 val state: StateFlow<ConnectionState> = _state.asStateFlow()
52 val events: SharedFlow<BusEvent> = stream()
53 // Buffer so a slow subscriber (e.g. a reducer under lock) can't suspend send()
54 // and stall the SSE read loop. Drop oldest on overflow — the next event carries
55 // the freshest state, so a dropped interim delta is harmless.
56 .buffer(bufferCapacity, BufferOverflow.DROP_OLDEST)
57 .shareIn(
58 scope = scope,
59 started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
60 replay = 0,
61 )
63 /** Conflated channel used to cut the reconnect backoff short when network returns. */
64 private val reconnectSignal = Channel<Unit>(Channel.CONFLATED)
66 /** Returns true if the exception is an SSE auth failure (401/403) that should stop retrying. */
67 private fun isSseAuthFailure(e: Throwable): Boolean =
68 e is io.ktor.client.plugins.ClientRequestException &&
69 (e.response.status.value == 401 || e.response.status.value == 403)
71 /** A log-safe exception summary that avoids leaking the request URL (which
72 * ClientRequestException embeds in its message and may contain auth or paths). */
73 private fun safeExceptionSummary(e: Throwable): String {
74 val status = (e as? io.ktor.client.plugins.ClientRequestException)?.response?.status?.value
75 return if (status != null) "${e.javaClass.simpleName}($status)" else e.javaClass.simpleName
76 }
78 /** Request an immediate reconnect, skipping any in-progress backoff. */
79 fun triggerReconnect() { reconnectSignal.trySend(Unit) }
81 /**
82 * Read SSE events from [events], racing each read against an idle-timeout watchdog.
83 * Throws [IOException] on idle timeout so the outer reconnect loop fires; breaks on
84 * clean stream close or scope cancellation.
85 */
86 private suspend fun readSseEvents(
87 scope: CoroutineScope,
88 events: kotlinx.coroutines.channels.ReceiveChannel<ServerSentEvent>,
89 send: suspend (BusEvent) -> Unit,
90 ) {
91 while (scope.isActive) {
92 // The OkHttp engine has readTimeout=0 (the SSE stream must never time out on
93 // read), so a socket the peer never closed would otherwise hang until OS TCP
94 // keepalive kicks in — minutes to hours on stock Linux. If nothing arrives
95 // within the idle timeout, treat the connection as half-open and drop it.
96 val result: ChannelResult<ServerSentEvent>? = select {
97 events.onReceiveCatching { it }
98 onTimeout(idleTimeoutMs) { null }
99 }
100 if (result == null) throw IOException("SSE idle timeout, reconnecting")
101 val sse = result.getOrNull() ?: break
102 val data = sse.data
103 if (data == null) {
104 // A comment or keep-alive event (no data field) resets the idle watchdog
105 // — servers send `: ping` comments to keep the connection alive through
106 // proxies. Without this, a server that only sends comments would still
107 // hit the idle timeout and reconnect unnecessarily.
108 continue
109 }
110 val event = try {
111 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
112 } catch (e: kotlinx.serialization.SerializationException) {
113 Log.d("EventStream", "Skipping unparseable SSE event", e)
114 continue
115 }
116 if (_state.value != ConnectionState.Connected) {
117 _state.value = ConnectionState.Connected
118 }
119 send(event)
120 }
121 }
123 private fun stream(): Flow<BusEvent> = channelFlow {
124 val scope = this
125 // Drain any stale reconnect signal from a previous collection cycle so it
126 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
127 // stopped the upstream while triggerReconnect() queued a signal).
128 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
129 var backoffMs = initialBackoffMs
130 while (isActive) {
131 _state.value = ConnectionState.Connecting
132 try {
133 // The SSE stream is long-lived: disable the request-level and socket
134 // timeouts (the client default is 60s for REST) in the request config
135 // so the connection isn't killed mid-stream. The idle-timeout watchdog
136 // below handles half-open connections instead.
137 client.sse(
138 "event",
139 request = {
140 timeout {
141 requestTimeoutMillis = Long.MAX_VALUE
142 // Disable the socket (read) timeout entirely for the SSE stream.
143 // A finite value would kill the connection whenever the server
144 // is quiet for longer than that (idle sessions, slow agent runs),
145 // causing a reconnect storm. Half-open detection is handled by
146 // the idle-timeout watchdog below plus OkHttp's pingInterval on
147 // HTTP/2 connections.
148 socketTimeoutMillis = Long.MAX_VALUE
149 }
150 },
151 ) {
152 backoffMs = initialBackoffMs
153 // The SSE block is open — the connection is established. Mark as
154 // Connected immediately so downstream re-seed logic fires even when
155 // the server only sends keep-alive comments (no data events).
156 if (_state.value != ConnectionState.Connected) {
157 _state.value = ConnectionState.Connected
158 }
159 // Clear any stale reconnect signal from a triggerReconnect() call
160 // made while the previous stream was healthy, so it doesn't suppress
161 // the next legitimate backoff.
162 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
163 // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
164 // element can be raced against an idle timeout via select.
165 val events = incoming.produceIn(scope)
166 try {
167 readSseEvents(scope, events) { send(it) }
168 } finally {
169 events.cancel()
170 }
171 }
172 } catch (c: CancellationException) {
173 throw c
174 } catch (e: Exception) {
175 // If the scope was cancelled (e.g. connection close), the closed-client
176 // exception is expected — don't log a spurious "stream error" warning.
177 if (isSseAuthFailure(e)) {
178 // Credentials are wrong — retrying won't help. Log a scrubbed
179 // summary (class + status) instead of the full exception, whose
180 // message carries the request URL and may include auth or paths.
181 Log.w("EventStream", "SSE auth failed (401/403), stopping retries: ${safeExceptionSummary(e)}")
182 _state.value = ConnectionState.Failed
183 break
184 } else if (isActive) {
185 Log.w("EventStream", "SSE stream error, will retry: ${safeExceptionSummary(e)}")
186 }
187 }
188 _state.value = ConnectionState.Disconnected
189 if (!isActive) break
190 // Wait for the backoff, but allow a reconnect signal to cut it short.
191 var signaled = false
192 val jitter = ((backoffMs * NetworkConfig.retryJitterFactor) * (Random.nextDouble() * 2 - 1)).toLong()
193 select {
194 reconnectSignal.onReceive { signaled = true }
195 onTimeout((backoffMs + jitter).coerceAtLeast(0)) { /* normal backoff elapsed */ }
196 }
197 backoffMs = if (signaled) initialBackoffMs else (backoffMs * 2).coerceAtMost(maxBackoffMs)
198 }
199 }