git.iko.soy/opencode-android
on 0bc1eda help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
137 lines 6.7 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
29/**
30 * Owns the single long-lived `GET /event` SSE subscription and exposes it as a hot
31 * [SharedFlow] of [BusEvent]. The connection is opened when the first collector
32 * subscribes (WhileSubscribed) and reconnects with backoff on drop.
33 *
34 * Ordering contract: callers that send a prompt must already be collecting [events]
35 * (directly or via the repository) so early `message.part.updated` deltas aren't missed.
36 */
37@OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class)
38class EventStreamClient(
39 private val client: HttpClient,
40 scope: CoroutineScope,
41) {
42 enum class ConnectionState { Disconnected, Connecting, Connected }
44 private val _state = MutableStateFlow(ConnectionState.Disconnected)
45 val state: StateFlow<ConnectionState> = _state.asStateFlow()
47 val events: SharedFlow<BusEvent> = stream()
48 // Buffer so a slow subscriber (e.g. a reducer under lock) can't suspend send()
49 // and stall the SSE read loop. Drop oldest on overflow — the next event carries
50 // the freshest state, so a dropped interim delta is harmless.
51 .buffer(NetworkConfig.sseEventBufferCapacity, BufferOverflow.DROP_OLDEST)
52 .shareIn(
53 scope = scope,
54 started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
55 replay = 0,
56 )
58 /** Conflated channel used to cut the reconnect backoff short when network returns. */
59 private val reconnectSignal = Channel<Unit>(Channel.CONFLATED)
61 /** Request an immediate reconnect, skipping any in-progress backoff. */
62 fun triggerReconnect() { reconnectSignal.trySend(Unit) }
64 private fun stream(): Flow<BusEvent> = channelFlow {
65 val scope = this
66 var backoffMs = NetworkConfig.sseInitialBackoffMs
67 while (isActive) {
68 _state.value = ConnectionState.Connecting
69 try {
70 // The SSE stream is long-lived: disable the request-level and socket
71 // timeouts (the client default is 60s for REST) in the request config
72 // so the connection isn't killed mid-stream. The idle-timeout watchdog
73 // below handles half-open connections instead.
74 client.sse(
75 "event",
76 request = {
77 timeout {
78 requestTimeoutMillis = Long.MAX_VALUE
79 socketTimeoutMillis = Long.MAX_VALUE
80 }
81 },
82 ) {
83 backoffMs = NetworkConfig.sseInitialBackoffMs
84 // `incoming` is a cold Flow; bridge it to a ReceiveChannel so each
85 // element can be raced against an idle timeout via select.
86 val events = incoming.produceIn(scope)
87 try {
88 while (isActive) {
89 // The OkHttp engine has readTimeout=0 (the SSE stream must
90 // never time out on read), so a socket the peer never closed
91 // would otherwise hang until OS TCP keepalive kicks in —
92 // minutes to hours on stock Linux. If nothing arrives within
93 // the idle timeout, treat the connection as half-open and
94 // drop it so the outer loop reconnects.
95 val result: ChannelResult<ServerSentEvent>? = select {
96 events.onReceiveCatching { it }
97 onTimeout(NetworkConfig.sseIdleTimeoutMs) { null }
98 }
99 // Timeout: drop and reconnect.
100 if (result == null) throw IOException("SSE idle timeout, reconnecting")
101 // Stream closed cleanly: exit to reconnect without a warning log.
102 val sse = result.getOrNull() ?: break
103 val data = sse.data ?: continue
104 val event = try {
105 OpencodeJson.decodeFromString(BusEvent.serializer(), data)
106 } catch (e: kotlinx.serialization.SerializationException) {
107 continue
108 }
109 // Any successfully decoded event means the stream is live;
110 // don't wait for a ServerConnected event to show "Connected".
111 if (_state.value != ConnectionState.Connected) {
112 _state.value = ConnectionState.Connected
113 }
114 send(event)
115 }
116 } finally {
117 events.cancel()
118 }
119 }
120 } catch (c: CancellationException) {
121 throw c
122 } catch (t: Throwable) {
123 Log.w("EventStream", "SSE stream error, will retry", t)
124 }
125 _state.value = ConnectionState.Disconnected
126 if (!isActive) break
127 // Wait for the backoff, but allow a reconnect signal to cut it short.
128 var signaled = false
129 val jitter = (backoffMs * 0.2 * Math.random()).toLong()
130 select {
131 reconnectSignal.onReceive { signaled = true }
132 onTimeout(backoffMs + jitter) { /* normal backoff elapsed */ }
133 }
134 backoffMs = if (signaled) NetworkConfig.sseInitialBackoffMs else (backoffMs * 2).coerceAtMost(NetworkConfig.sseMaxBackoffMs)
135 }
136 }