git.iko.soy/opencode-android
on 3e723a1 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
160 lines 8.3 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 private fun stream(): Flow<BusEvent> = channelFlow {
71 val scope = this
72 // Drain any stale reconnect signal from a previous collection cycle so it
73 // doesn't spuriously reset the backoff on startup (WhileSubscribed may have
74 // stopped the upstream while triggerReconnect() queued a signal).
75 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
76 var backoffMs = NetworkConfig.sseInitialBackoffMs
77 while (isActive) {
78 _state.value = ConnectionState.Connecting
79 try {
80 // The SSE stream is long-lived: disable the request-level and socket
81 // timeouts (the client default is 60s for REST) in the request config
82 // so the connection isn't killed mid-stream. The idle-timeout watchdog
83 // below handles half-open connections instead.
84 client.sse(
85 "event",
86 request = {
87 timeout {
88 requestTimeoutMillis = Long.MAX_VALUE
89 // Use a finite socket timeout so a server that accepts TCP but
90 // never sends SSE headers doesn't hang indefinitely. Once events
91 // are flowing, the idle-timeout watchdog handles half-open detection.
92 socketTimeoutMillis = NetworkConfig.sseHandshakeTimeoutMs
93 }
94 },
95 ) {
96 backoffMs = NetworkConfig.sseInitialBackoffMs
97 // Clear any stale reconnect signal from a triggerReconnect() call
98 // made while the previous stream was healthy, so it doesn't suppress
99 // the next legitimate backoff.
100 while (reconnectSignal.tryReceive().isSuccess) { /* drain */ }
101 // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
102 // element can be raced against an idle timeout via select.
103 val events = incoming.produceIn(scope)
104 try {
105 while (isActive) {
106 // The OkHttp engine has readTimeout=0 (the SSE stream must
107 // never time out on read), so a socket the peer never closed
108 // would otherwise hang until OS TCP keepalive kicks in —
109 // minutes to hours on stock Linux. If nothing arrives within
110 // the idle timeout, treat the connection as half-open and
111 // drop it so the outer loop reconnects.
112 val result: ChannelResult<ServerSentEvent>? = select {
113 events.onReceiveCatching { it }
114 onTimeout(NetworkConfig.sseIdleTimeoutMs) { null }
115 }
116 // Timeout: drop and reconnect.
117 if (result == null) throw IOException("SSE idle timeout, reconnecting")
118 // Stream closed cleanly: exit to reconnect without a warning log.
119 val sse = result.getOrNull() ?: break
120 val data = sse.data ?: continue
121 val event = try {
122 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
123 } catch (e: kotlinx.serialization.SerializationException) {
124 Log.d("EventStream", "Skipping unparseable SSE event", e)
125 continue
126 }
127 // Any successfully decoded event means the stream is live;
128 // don't wait for a ServerConnected event to show "Connected".
129 if (_state.value != ConnectionState.Connected) {
130 _state.value = ConnectionState.Connected
131 }
132 send(event)
133 }
134 } finally {
135 events.cancel()
136 }
137 }
138 } catch (c: CancellationException) {
139 throw c
140 } catch (e: Exception) {
141 if (isSseAuthFailure(e)) {
142 Log.w("EventStream", "SSE auth failed, stopping reconnect", e)
143 _state.value = ConnectionState.Disconnected
144 throw e
145 }
146 Log.w("EventStream", "SSE stream error, will retry", e)
147 }
148 _state.value = ConnectionState.Disconnected
149 if (!isActive) break
150 // Wait for the backoff, but allow a reconnect signal to cut it short.
151 var signaled = false
152 val jitter = (backoffMs * 0.2 * Random.nextDouble()).toLong()
153 select {
154 reconnectSignal.onReceive { signaled = true }
155 onTimeout(backoffMs + jitter) { /* normal backoff elapsed */ }
156 }
157 backoffMs = if (signaled) NetworkConfig.sseInitialBackoffMs else (backoffMs * 2).coerceAtMost(NetworkConfig.sseMaxBackoffMs)
158 }
159 }