git.iko.soy/opencode-android
on e37d1bb help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
78 lines 3.1 kB history raw
1package soy.iko.opencode.data.network
3import soy.iko.opencode.data.model.BusEvent
4import io.ktor.client.HttpClient
5import io.ktor.client.plugins.sse.sse
6import kotlinx.coroutines.CancellationException
7import kotlinx.coroutines.CoroutineScope
8import kotlinx.coroutines.delay
9import kotlinx.coroutines.flow.Flow
10import kotlinx.coroutines.flow.MutableStateFlow
11import kotlinx.coroutines.flow.SharedFlow
12import kotlinx.coroutines.flow.StateFlow
13import kotlinx.coroutines.flow.SharingStarted
14import kotlinx.coroutines.flow.asStateFlow
15import kotlinx.coroutines.flow.channelFlow
16import kotlinx.coroutines.flow.shareIn
17import kotlinx.coroutines.isActive
19/**
20 * Owns the single long-lived `GET /event` SSE subscription and exposes it as a hot
21 * [SharedFlow] of [BusEvent]. The connection is opened when the first collector
22 * subscribes (WhileSubscribed) and reconnects with backoff on drop.
23 *
24 * Ordering contract: callers that send a prompt must already be collecting [events]
25 * (directly or via the repository) so early `message.part.updated` deltas aren't missed.
26 */
27class EventStreamClient(
28 private val client: HttpClient,
29 scope: CoroutineScope,
30) {
31 enum class ConnectionState { Disconnected, Connecting, Connected }
33 private val _state = MutableStateFlow(ConnectionState.Disconnected)
34 val state: StateFlow<ConnectionState> = _state.asStateFlow()
36 val events: SharedFlow<BusEvent> = stream()
37 .shareIn(
38 scope = scope,
39 started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
40 replay = 0,
41 )
43 private fun stream(): Flow<BusEvent> = channelFlow {
44 var backoffMs = INITIAL_BACKOFF_MS
45 while (isActive) {
46 _state.value = ConnectionState.Connecting
47 try {
48 client.sse("event") {
49 backoffMs = INITIAL_BACKOFF_MS
50 incoming.collect { sse ->
51 val data = sse.data ?: return@collect
52 val event = runCatching { OpencodeJson.decodeFromString(BusEvent.serializer(), data) }
53 .getOrNull() ?: return@collect
54 // Any successfully decoded event means the stream is live;
55 // don't wait for a ServerConnected event to show "Connected".
56 if (_state.value != ConnectionState.Connected) {
57 _state.value = ConnectionState.Connected
58 }
59 send(event)
60 }
61 }
62 } catch (c: CancellationException) {
63 throw c
64 } catch (t: Throwable) {
65 // transient error — fall through to reconnect
66 }
67 _state.value = ConnectionState.Disconnected
68 if (!isActive) break
69 delay(backoffMs)
70 backoffMs = (backoffMs * 2).coerceAtMost(MAX_BACKOFF_MS)
71 }
72 }
74 private companion object {
75 const val INITIAL_BACKOFF_MS = 500L
76 const val MAX_BACKOFF_MS = 10_000L
77 }