git.iko.soy/opencode-android
on f9ed438 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt
75 lines 2.8 kB history raw
1package soy.iko.opencode.data.network
3import soy.iko.opencode.data.model.BusEvent
4import soy.iko.opencode.data.model.ServerConnected
5import io.ktor.client.HttpClient
6import io.ktor.client.plugins.sse.sse
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 if (event is ServerConnected) {
55 _state.value = ConnectionState.Connected
56 }
57 send(event)
58 }
59 }
60 } catch (t: Throwable) {
61 if (!isActive) throw t
62 // fall through to reconnect
63 }
64 _state.value = ConnectionState.Disconnected
65 if (!isActive) break
66 delay(backoffMs)
67 backoffMs = (backoffMs * 2).coerceAtMost(MAX_BACKOFF_MS)
68 }
69 }
71 private companion object {
72 const val INITIAL_BACKOFF_MS = 500L
73 const val MAX_BACKOFF_MS = 10_000L
74 }