feat(android): EventsStream — SSE subscription to /api/events/stream
Foundation for audit #4 (cross-device reactivity). Long-lived SSE subscription exposed as a process-wide SharedFlow<LiveEvent>; gated on having a session cookie (opens on sign-in, closes on sign-out). Mirrors flutter_client/lib/shared/live_events_provider.dart in behavior — no client-side timeout (server heartbeats every 15s), no explicit reconnect-with-backoff in v1 (auth transitions re-open). LiveEvent carries kind / userId / data (JsonObject); consumers deserialize the payload per event kind they handle. Force-injected into MinstrelApplication via the same pattern as MutationReplayer / SyncController / ResumeController / PlayEventsReporter so the singleton constructs at app start. okhttp-sse was already on the classpath; no new deps. No consumers yet — LiveEventsDispatcher + per-screen subscribers land in follow-up commits. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -9,6 +9,7 @@ import coil3.network.okhttp.OkHttpNetworkFetcherFactory
|
|||||||
import com.fabledsword.minstrel.cache.mutations.MutationReplayer
|
import com.fabledsword.minstrel.cache.mutations.MutationReplayer
|
||||||
import com.fabledsword.minstrel.cache.sync.SyncController
|
import com.fabledsword.minstrel.cache.sync.SyncController
|
||||||
import com.fabledsword.minstrel.di.ApplicationScope
|
import com.fabledsword.minstrel.di.ApplicationScope
|
||||||
|
import com.fabledsword.minstrel.events.EventsStream
|
||||||
import com.fabledsword.minstrel.player.PlayEventsReporter
|
import com.fabledsword.minstrel.player.PlayEventsReporter
|
||||||
import com.fabledsword.minstrel.player.ResumeController
|
import com.fabledsword.minstrel.player.ResumeController
|
||||||
import dagger.hilt.android.HiltAndroidApp
|
import dagger.hilt.android.HiltAndroidApp
|
||||||
@@ -68,6 +69,15 @@ class MinstrelApplication :
|
|||||||
*/
|
*/
|
||||||
@Suppress("unused") @Inject lateinit var playEventsReporter: PlayEventsReporter
|
@Suppress("unused") @Inject lateinit var playEventsReporter: PlayEventsReporter
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Same construct-the-singleton trick — EventsStream's init block
|
||||||
|
* subscribes to AuthStore.sessionCookie and opens / closes the
|
||||||
|
* `/api/events/stream` SSE subscription on auth transitions.
|
||||||
|
* Without this @Inject the events SharedFlow would have no
|
||||||
|
* upstream and consumers would never see anything.
|
||||||
|
*/
|
||||||
|
@Suppress("unused") @Inject lateinit var eventsStream: EventsStream
|
||||||
|
|
||||||
@Inject @ApplicationScope lateinit var appScope: CoroutineScope
|
@Inject @ApplicationScope lateinit var appScope: CoroutineScope
|
||||||
|
|
||||||
override fun onCreate() {
|
override fun onCreate() {
|
||||||
|
|||||||
@@ -0,0 +1,141 @@
|
|||||||
|
package com.fabledsword.minstrel.events
|
||||||
|
|
||||||
|
import com.fabledsword.minstrel.auth.AuthStore
|
||||||
|
import com.fabledsword.minstrel.di.ApplicationScope
|
||||||
|
import kotlinx.coroutines.CoroutineScope
|
||||||
|
import kotlinx.coroutines.channels.BufferOverflow
|
||||||
|
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||||
|
import kotlinx.coroutines.flow.SharedFlow
|
||||||
|
import kotlinx.coroutines.flow.asSharedFlow
|
||||||
|
import kotlinx.coroutines.flow.distinctUntilChanged
|
||||||
|
import kotlinx.coroutines.flow.map
|
||||||
|
import kotlinx.coroutines.launch
|
||||||
|
import kotlinx.serialization.json.Json
|
||||||
|
import kotlinx.serialization.json.JsonObject
|
||||||
|
import kotlinx.serialization.json.contentOrNull
|
||||||
|
import kotlinx.serialization.json.jsonObject
|
||||||
|
import kotlinx.serialization.json.jsonPrimitive
|
||||||
|
import okhttp3.OkHttpClient
|
||||||
|
import okhttp3.Request
|
||||||
|
import okhttp3.Response
|
||||||
|
import okhttp3.sse.EventSource
|
||||||
|
import okhttp3.sse.EventSourceListener
|
||||||
|
import okhttp3.sse.EventSources
|
||||||
|
import javax.inject.Inject
|
||||||
|
import javax.inject.Singleton
|
||||||
|
|
||||||
|
private const val SSE_PATH = "/api/events/stream"
|
||||||
|
private const val EVENTS_BUFFER_CAPACITY = 64
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Long-lived SSE subscription to `GET /api/events/stream`. Exposes
|
||||||
|
* [events] as a process-wide [SharedFlow]; consumers (per-screen
|
||||||
|
* ViewModels + the central [LiveEventsDispatcher]) collect filtered
|
||||||
|
* subsets of the stream.
|
||||||
|
*
|
||||||
|
* Connection lifecycle mirrors
|
||||||
|
* `flutter_client/lib/shared/live_events_provider.dart`:
|
||||||
|
* - Gated on having a session cookie. Subscription opens when the
|
||||||
|
* cookie transitions to non-null and closes when it transitions
|
||||||
|
* back to null (sign-out).
|
||||||
|
* - No client-side timeout — the server emits 15s heartbeats which
|
||||||
|
* okhttp-sse handles transparently.
|
||||||
|
* - No explicit reconnect-with-backoff in v1; if the stream drops
|
||||||
|
* mid-session (server restart, network blip) it stays closed until
|
||||||
|
* the next auth-cookie transition. Matches Flutter's behavior.
|
||||||
|
*
|
||||||
|
* The URL passes through the placeholder host that
|
||||||
|
* `BaseUrlInterceptor` rewrites — same mechanism the rest of the
|
||||||
|
* Retrofit traffic uses, so server-URL changes propagate without
|
||||||
|
* needing a separate path through here.
|
||||||
|
*/
|
||||||
|
@Singleton
|
||||||
|
class EventsStream @Inject constructor(
|
||||||
|
private val authStore: AuthStore,
|
||||||
|
@ApplicationScope private val scope: CoroutineScope,
|
||||||
|
private val okHttpClient: OkHttpClient,
|
||||||
|
private val json: Json,
|
||||||
|
) {
|
||||||
|
private val factory = EventSources.createFactory(okHttpClient)
|
||||||
|
|
||||||
|
private val emitter = MutableSharedFlow<LiveEvent>(
|
||||||
|
replay = 0,
|
||||||
|
extraBufferCapacity = EVENTS_BUFFER_CAPACITY,
|
||||||
|
onBufferOverflow = BufferOverflow.DROP_OLDEST,
|
||||||
|
)
|
||||||
|
val events: SharedFlow<LiveEvent> = emitter.asSharedFlow()
|
||||||
|
|
||||||
|
private var currentSource: EventSource? = null
|
||||||
|
|
||||||
|
init {
|
||||||
|
scope.launch {
|
||||||
|
authStore.sessionCookie
|
||||||
|
.map { !it.isNullOrEmpty() }
|
||||||
|
.distinctUntilChanged()
|
||||||
|
.collect { signedIn ->
|
||||||
|
currentSource?.cancel()
|
||||||
|
currentSource = if (signedIn) openSubscription() else null
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun openSubscription(): EventSource {
|
||||||
|
val request = Request.Builder()
|
||||||
|
.url("http://placeholder.invalid$SSE_PATH")
|
||||||
|
.header("Accept", "text/event-stream")
|
||||||
|
.build()
|
||||||
|
return factory.newEventSource(request, Listener())
|
||||||
|
}
|
||||||
|
|
||||||
|
private inner class Listener : EventSourceListener() {
|
||||||
|
override fun onEvent(
|
||||||
|
eventSource: EventSource,
|
||||||
|
id: String?,
|
||||||
|
type: String?,
|
||||||
|
data: String,
|
||||||
|
) {
|
||||||
|
parseFrame(kind = type, data = data)?.let(emitter::tryEmit)
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun onClosed(eventSource: EventSource) {
|
||||||
|
// Stream ended (server-side close or network drop). No
|
||||||
|
// explicit reconnect — auth-cookie transitions re-open.
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun onFailure(
|
||||||
|
eventSource: EventSource,
|
||||||
|
t: Throwable?,
|
||||||
|
response: Response?,
|
||||||
|
) {
|
||||||
|
// Same as onClosed — auth transition will re-open. The
|
||||||
|
// throwable/response carry diagnostics; we don't surface
|
||||||
|
// them in v1 since the stream is "best effort" by design.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Parses one SSE frame into a [LiveEvent]. The SSE `event:` field
|
||||||
|
* carries the kind; the `data:` field is a JSON object with
|
||||||
|
* `kind`/`user_id`/`data` (the kind in the payload is redundant
|
||||||
|
* with the SSE event field — we prefer the SSE field, falling
|
||||||
|
* back to the payload field). Heartbeat comment frames don't
|
||||||
|
* reach here (okhttp-sse filters them).
|
||||||
|
*/
|
||||||
|
private fun parseFrame(kind: String?, data: String): LiveEvent? = try {
|
||||||
|
val payload = json.parseToJsonElement(data).jsonObject
|
||||||
|
val resolvedKind = kind ?: payload["kind"]?.jsonPrimitive?.contentOrNull
|
||||||
|
if (resolvedKind.isNullOrEmpty()) {
|
||||||
|
null
|
||||||
|
} else {
|
||||||
|
LiveEvent(
|
||||||
|
kind = resolvedKind,
|
||||||
|
userId = payload["user_id"]?.jsonPrimitive?.contentOrNull.orEmpty(),
|
||||||
|
data = (payload["data"] as? JsonObject) ?: JsonObject(emptyMap()),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
} catch (
|
||||||
|
@Suppress("TooGenericExceptionCaught") _: Throwable,
|
||||||
|
) {
|
||||||
|
null
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
package com.fabledsword.minstrel.events
|
||||||
|
|
||||||
|
import kotlinx.serialization.Serializable
|
||||||
|
import kotlinx.serialization.json.JsonObject
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Parsed event from the server's SSE stream. Mirrors
|
||||||
|
* `flutter_client/lib/shared/live_events_provider.dart`'s `LiveEvent`.
|
||||||
|
*
|
||||||
|
* - [kind] is the SSE `event:` field (e.g. "track.liked", "playlist.deleted").
|
||||||
|
* - [userId] is the actor whose user-scoped state changed (empty for
|
||||||
|
* broadcast events like scan.* that aren't user-bound).
|
||||||
|
* - [data] is the raw event payload; consumers deserialize into the
|
||||||
|
* typed shape they expect for the kind they handle.
|
||||||
|
*/
|
||||||
|
@Serializable
|
||||||
|
data class LiveEvent(
|
||||||
|
val kind: String,
|
||||||
|
val userId: String,
|
||||||
|
val data: JsonObject,
|
||||||
|
)
|
||||||
Reference in New Issue
Block a user