RawCastObserver.kt
1255 lines · 64.0 KB
Changed in: 0.5.0
| 1 | package dev.stave.cast.raw |
| 2 | |
| 3 | import dev.stave.cast.CastDevice |
| 4 | import dev.stave.cast.CastIntegration |
| 5 | import dev.stave.data.SettingsStore |
| 6 | import dev.stave.domain.model.DidlItem |
| 7 | import dev.stave.domain.renderer.PlaybackState |
| 8 | import kotlinx.coroutines.CancellationException |
| 9 | import kotlinx.coroutines.CoroutineScope |
| 10 | import kotlinx.coroutines.Dispatchers |
| 11 | import kotlinx.coroutines.Job |
| 12 | import kotlinx.coroutines.SupervisorJob |
| 13 | import kotlinx.coroutines.cancel |
| 14 | import kotlinx.coroutines.delay |
| 15 | import kotlinx.coroutines.flow.MutableStateFlow |
| 16 | import kotlinx.coroutines.flow.StateFlow |
| 17 | import kotlinx.coroutines.flow.asStateFlow |
| 18 | import kotlinx.coroutines.flow.combine |
| 19 | import kotlinx.coroutines.flow.distinctUntilChanged |
| 20 | import kotlinx.coroutines.flow.drop |
| 21 | import kotlinx.coroutines.flow.first |
| 22 | import kotlinx.coroutines.flow.map |
| 23 | import kotlinx.coroutines.flow.update |
| 24 | import kotlinx.coroutines.isActive |
| 25 | import kotlinx.coroutines.withTimeoutOrNull |
| 26 | import kotlinx.coroutines.launch |
| 27 | import kotlinx.coroutines.sync.Mutex |
| 28 | import kotlinx.coroutines.sync.withLock |
| 29 | import kotlin.time.Duration |
| 30 | import kotlin.time.Duration.Companion.seconds |
| 31 | |
| 32 | /** |
| 33 | * Watches a Chromecast/CCA-group session driven by some *other* sender |
| 34 | * (typically the Earhorn phone app) and exposes its state as a [PlaybackState] |
| 35 | * flow the desktop UI can render. |
| 36 | * |
| 37 | * Discovery is delegated to [CastIntegration] — it already runs a JmDNS scan for |
| 38 | * `_googlecast._tcp.` services and exposes a [CastIntegration.devices] flow. The |
| 39 | * picker passes the user-selected entry to [observe], we pull `host` and `port` |
| 40 | * off the [CastDevice], and open a raw TLS connection via [RawCastClient]. No |
| 41 | * parallel discovery stack. |
| 42 | * |
| 43 | * Two distinct stacks for the desktop: |
| 44 | * - **Sender** (existing): [CastIntegration] + `CastRenderer` + `CastStreamProxy`. |
| 45 | * Driven by the [RendererCoordinator]'s active delegate. Used when the user |
| 46 | * wants the desktop itself to *cast to* a group. |
| 47 | * - **Observer** (this class): [RawCastClient] over its own TLS socket. Does |
| 48 | * NOT touch the coordinator's active delegate. Used when the user wants the |
| 49 | * desktop to *follow* a group that another sender is driving. |
| 50 | * |
| 51 | * They are mutually exclusive in the UI (a single "is the user observing?" |
| 52 | * flag, [observing]) but architecturally independent — calling [observe] does |
| 53 | * not disturb the renderer pipeline. |
| 54 | * |
| 55 | * Ported from the Android TV implementation. The Android version did its own |
| 56 | * mDNS via `NsdManager` and held a `MulticastLock`; both are unnecessary on |
| 57 | * desktop where the JVM mDNS layer just works and we have a working discovery |
| 58 | * pipeline already. |
| 59 | * |
| 60 | * **Follow survives restart.** The followed device id is persisted |
| 61 | * ([SettingsStore.followedCastId]) on every [observe] and cleared only by |
| 62 | * [disconnect] — which is exclusively user-initiated ("stop following" / |
| 63 | * eyeball toggle). At construction, [restorePersistedFollow] waits for the |
| 64 | * persisted device to show up in discovery and resumes the follow, unless the |
| 65 | * user does something renderer-related first (picks a renderer, follows |
| 66 | * something else, or stops). This is the deliberate mirror image of the |
| 67 | * sender-side policy: `cast:` sender selections are dropped at settings load |
| 68 | * so a launch never auto-ATTACHES to a group (attaching grabs the transport), |
| 69 | * while a follow — being a passive overlay — is safe and useful to resume. |
| 70 | */ |
| 71 | class RawCastObserver( |
| 72 | /** Reused for discovery. We don't drive the sender stack from here; the |
| 73 | * observer is orthogonal. */ |
| 74 | private val castIntegration: CastIntegration, |
| 75 | /** musicnow's /status, for the song a radio station is playing. */ |
| 76 | private val radio: dev.stave.cast.radio.RadioStatusClient, |
| 77 | /** Persistence for the follow-across-restarts contract (see class KDoc) — |
| 78 | * reads/writes [SettingsStore.followedCastId] and reads |
| 79 | * [SettingsStore.activeRendererUdn] as the "user picked a renderer this |
| 80 | * session" supersede signal for the startup restore. */ |
| 81 | private val settings: SettingsStore, |
| 82 | ) : CastObserverApi { |
| 83 | /** Devices visible on the LAN that the user can pick to follow. Pass-through |
| 84 | * from [CastIntegration.devices] — saves a duplicate JmDNS stack and keeps |
| 85 | * the picker showing one list regardless of which mode the user picks. */ |
| 86 | val devices: StateFlow<List<CastDevice>> = castIntegration.devices |
| 87 | |
| 88 | private val _playbackState = MutableStateFlow(PlaybackState()) |
| 89 | /** The station overlay from musicnow, or null. Applied on top of the |
| 90 | * receiver's static station title in [playbackState] (the maintainer's |
| 91 | * answer 2, 2026-09-04). */ |
| 92 | @Volatile private var radioNow: dev.stave.cast.radio.RadioStatusClient.RadioNow? = null |
| 93 | /** Playback state mapped from raw [RawCastClient.MediaStatus] into the |
| 94 | * domain [PlaybackState] shape. Reads identical to `renderer.state` from |
| 95 | * the UI's perspective. */ |
| 96 | override val playbackState: StateFlow<PlaybackState> = _playbackState.asStateFlow() |
| 97 | |
| 98 | private val _observing = MutableStateFlow(false) |
| 99 | /** True while connected to a Cast device in observer mode. UI surfaces gate |
| 100 | * their `tvTrack` / `tvState` resolution on this flag. */ |
| 101 | override val observing: StateFlow<Boolean> = _observing.asStateFlow() |
| 102 | |
| 103 | private val _connectedDeviceName = MutableStateFlow<String?>(null) |
| 104 | /** Friendly name of the currently-observed device, or null when not |
| 105 | * observing. Used by the picker to show the active follow target. */ |
| 106 | override val connectedDeviceName: StateFlow<String?> = _connectedDeviceName.asStateFlow() |
| 107 | |
| 108 | private val _sessionLive = MutableStateFlow(false) |
| 109 | /** See [CastObserverApi.sessionLive]. Raw: recomputed on every status, |
| 110 | * dropped by the silence watchdog and the frozen-position latch. */ |
| 111 | override val sessionLive: StateFlow<Boolean> = _sessionLive.asStateFlow() |
| 112 | |
| 113 | private val _pendingTargetName = MutableStateFlow<String?>(null) |
| 114 | /** Friendly name of the device a manual follow pick ([observeOrRestore]) |
| 115 | * is still connecting to, or null. The picker shows "Connecting…" on |
| 116 | * that row and ignores further picks until the outcome (ported from |
| 117 | * Android 0.4.21/0.4.22). Bounded by [ATTACH_TIMEOUT_MS]. */ |
| 118 | val pendingTargetName: StateFlow<String?> = _pendingTargetName.asStateFlow() |
| 119 | |
| 120 | @Volatile private var client: RawCastClient? = null |
| 121 | @Volatile private var scope: CoroutineScope? = null |
| 122 | |
| 123 | /** Serializes session bring-up and session-replacing teardowns: a startup |
| 124 | * restore, a manual pick and a leader move used to interleave their |
| 125 | * writes to [client]/[scope] around the suspending connect(), leaking a |
| 126 | * session whose collectors kept writing [playbackState]. */ |
| 127 | private val sessionMutex = Mutex() |
| 128 | /** Bumped by every [teardownSession]. A bring-up that finds it changed |
| 129 | * after its connect() was superseded (the user stopped following |
| 130 | * meanwhile — [disconnect] is non-suspending, so it can't take the lock) |
| 131 | * and must not publish its session. */ |
| 132 | private val sessionGen = java.util.concurrent.atomic.AtomicInteger() |
| 133 | |
| 134 | /** Outcome of one bring-up. [Superseded] is not a failure: the user (or a |
| 135 | * newer request) ended this attempt, so nothing is reported. */ |
| 136 | private enum class AttachOutcome { Attached, Superseded, Failed } |
| 137 | |
| 138 | // ── Session liveness (Android F2 on the glass + freeze latch) ───── |
| 139 | /** The receiver claims PLAYING/BUFFERING but its position didn't move |
| 140 | * between two consecutive reports. The bar holds instead of advancing |
| 141 | * over silence. */ |
| 142 | @Volatile private var positionFrozen = false |
| 143 | @Volatile private var frozenSinceMs = 0L |
| 144 | /** A freeze that outlasted [ZOMBIE_FREEZE_MS]: the session is dead despite |
| 145 | * its PLAYING claim (Classical KC overnight, 2026-08-30). Withdrawn |
| 146 | * when the position moves again. */ |
| 147 | @Volatile private var zombieDeclared = false |
| 148 | /** No playback report for [STATUS_SILENCE_MS]. Withdrawn by the next one. */ |
| 149 | @Volatile private var silenceDeclared = false |
| 150 | |
| 151 | /** Lifetime scope for cross-session concerns: the startup follow-restore |
| 152 | * watcher and persistence writes from the non-suspend [disconnect]. Never |
| 153 | * cancelled — this is a process-scoped singleton, and [scope] (which dies |
| 154 | * with each observation) is deliberately NOT used for these, because a |
| 155 | * session teardown must not abort the persistence write recording it. */ |
| 156 | private val lifecycleScope = CoroutineScope(SupervisorJob() + Dispatchers.IO) |
| 157 | |
| 158 | /** The one-shot startup restore (see [restorePersistedFollow]). Cancelled by |
| 159 | * any manual [observe] or [disconnect] — a user action supersedes it. */ |
| 160 | private var restoreJob: Job? = null |
| 161 | |
| 162 | // ── Follower-queue state ───────────────────────────────────── |
| 163 | // |
| 164 | // The receiver's queue, assembled from the wire traffic RawCastClient |
| 165 | // surfaces as QueueEvents. Mutated ONLY inside the single serialized |
| 166 | // queue-event collector in [observe] (so no lock is needed for writes); |
| 167 | // published as immutable @Volatile snapshots so the mediaStatus bridge |
| 168 | // and [jumpToQueueItem] can read consistently from any thread. |
| 169 | |
| 170 | /** Receiver's ordered item-id list — the QUEUE_ITEM_IDS response. */ |
| 171 | @Volatile private var queueOrder: List<Int> = emptyList() |
| 172 | |
| 173 | /** Fetched metadata per itemId. Survivors are retained across refetches |
| 174 | * (per the shared spec) so the list doesn't flash empty on QUEUE_CHANGE. */ |
| 175 | @Volatile private var queueItemsById: Map<Int, DidlItem.Track> = emptyMap() |
| 176 | |
| 177 | /** Debounce for QUEUE_CHANGE-triggered refetches — coalesces the burst a |
| 178 | * driving sender produces during a bulk edit into one id-list round trip. */ |
| 179 | private var queueChangeDebounce: kotlinx.coroutines.Job? = null |
| 180 | |
| 181 | /** Fail-safe: if the device hasn't volunteered a volume reading within |
| 182 | * the bootstrap window, force the [setVolume] gate open anyway so the |
| 183 | * user gets back the ability to drive the slider. The first successful |
| 184 | * SET_VOLUME will likely trigger a RECEIVER_STATUS reply that includes |
| 185 | * the volume sub-object, latching volumeKnown for real on the client |
| 186 | * side; this is just here to make sure we don't strand the user on |
| 187 | * devices that never broadcast volume on their own initiative. */ |
| 188 | private val _forceVolumeInputAllowed = MutableStateFlow(false) |
| 189 | |
| 190 | /** Wakes the musicnow poller: session content changed, the musicnow |
| 191 | * address changed, or the user asked for a refresh. Declared before |
| 192 | * `init` so the startup follow-restore can never see it uninitialized. Conflated — several wakes before the loop runs are one. */ |
| 193 | private val radioWake = kotlinx.coroutines.channels.Channel<Unit>(kotlinx.coroutines.channels.Channel.CONFLATED) |
| 194 | |
| 195 | |
| 196 | /** Name of a group that could not be reached, for the UI to show once. */ |
| 197 | val lastFailure = MutableStateFlow<String?>(null) |
| 198 | fun consumeFailure() { lastFailure.value = null } |
| 199 | |
| 200 | init { |
| 201 | restoreJob = lifecycleScope.launch { restorePersistedFollow() } |
| 202 | } |
| 203 | |
| 204 | // ── Startup follow restore ────────────────────────────────── |
| 205 | |
| 206 | /** |
| 207 | * If a follow was active when the app last ran, resume it. One-shot per |
| 208 | * launch: waits (indefinitely — a speaker that's powered off tonight may |
| 209 | * be on tomorrow, and this coroutine is the only thing waiting) until the |
| 210 | * persisted device appears in discovery, then re-observes it. Because |
| 211 | * Follow is a passive overlay, resuming at launch is safe: it reads the |
| 212 | * group's session without touching its transport — unlike a sender attach, |
| 213 | * which is exactly what startup must never do. |
| 214 | * |
| 215 | * The wait aborts, leaving the persisted id in place only where noted, if |
| 216 | * the user acts first: |
| 217 | * - a manual [observe] or [disconnect] cancels [restoreJob] outright |
| 218 | * (observe re-persists its own target; disconnect clears); |
| 219 | * - the persisted follow id changing under us (shouldn't happen from |
| 220 | * another writer today, but the check makes the watcher self-consistent) |
| 221 | * aborts; |
| 222 | * - the user picking any renderer this session aborts WITHOUT clearing |
| 223 | * the persisted id. Rationale: they launched the app and immediately |
| 224 | * started driving something — having the UI flip to "Following X" |
| 225 | * thirty seconds later because X finally answered mDNS would be the |
| 226 | * same kind of startup ambush this feature set exists to kill. The |
| 227 | * follow still restores on the NEXT launch if they leave it be. |
| 228 | * [SettingsStore.activeRendererUdn] is compared against its value at |
| 229 | * launch rather than against null so that a persisted UPnP renderer |
| 230 | * selection (which legitimately survives restart) doesn't count as a |
| 231 | * user action. |
| 232 | * |
| 233 | * A failed re-observe (device answered mDNS but refused TLS, mid-reboot, |
| 234 | * etc.) logs, resets session state, and keeps the persisted id so the next |
| 235 | * launch tries again. |
| 236 | * |
| 237 | * Last-known-host rescue (Android parity): a freshly rebooted router can |
| 238 | * sit multicast-dead for minutes, so mDNS fails exactly when a cold start |
| 239 | * needs it. If discovery hasn't found the device after |
| 240 | * [LAST_KNOWN_HOST_GRACE_MS], dial the host it last attached at |
| 241 | * ([SettingsStore.lastKnownHosts]) once, through [observe]. If the group's |
| 242 | * leader has since moved, the moved-leader handling corrects it once |
| 243 | * discovery sees the group. If the dial fails, the persisted id stays and |
| 244 | * the wait for discovery carries on. |
| 245 | */ |
| 246 | private suspend fun restorePersistedFollow() { |
| 247 | val target = settings.followedCastId.value ?: return |
| 248 | val rendererAtLaunch = settings.activeRendererUdn.value |
| 249 | System.err.println( |
| 250 | "[EarhornCast/RawObserver] persisted follow for '$target' — " + |
| 251 | "will resume once the device is discovered", |
| 252 | ) |
| 253 | val probes = combine( |
| 254 | castIntegration.devices, |
| 255 | settings.followedCastId, |
| 256 | settings.activeRendererUdn, |
| 257 | ) { devs, follow, rendererNow -> |
| 258 | when { |
| 259 | follow != target -> RestoreProbe.Superseded |
| 260 | rendererNow != rendererAtLaunch -> RestoreProbe.Superseded |
| 261 | else -> { |
| 262 | val found = devs.firstOrNull { it.id == target } |
| 263 | if (found != null) RestoreProbe.Found(found) else RestoreProbe.Waiting |
| 264 | } |
| 265 | } |
| 266 | } |
| 267 | val probe = withTimeoutOrNull(LAST_KNOWN_HOST_GRACE_MS) { |
| 268 | probes.first { it !is RestoreProbe.Waiting } |
| 269 | } ?: run { |
| 270 | if (rescueByLastKnownHost(target, rendererAtLaunch)) return |
| 271 | probes.first { it !is RestoreProbe.Waiting } |
| 272 | } |
| 273 | |
| 274 | val device = (probe as? RestoreProbe.Found)?.device ?: run { |
| 275 | System.err.println( |
| 276 | "[EarhornCast/RawObserver] follow restore for '$target' " + |
| 277 | "superseded by a user action this session", |
| 278 | ) |
| 279 | return |
| 280 | } |
| 281 | if (_observing.value) return // user beat us to it between emissions |
| 282 | |
| 283 | // observe() cancels restoreJob as its first act (a manual pick must |
| 284 | // supersede a pending restore). We're INSIDE restoreJob right now, so |
| 285 | // null the handle first — otherwise observe() would cancel the very |
| 286 | // coroutine running it, and the restore would die mid-flight at its |
| 287 | // next suspension point. |
| 288 | restoreJob = null |
| 289 | try { |
| 290 | observe(device) |
| 291 | } catch (e: CancellationException) { |
| 292 | throw e |
| 293 | } catch (e: Exception) { |
| 294 | // observe() has already reset its half-initialized session |
| 295 | // (teardownSession, not disconnect, so the persisted id stays). |
| 296 | // Tearing down again here, outside the session lock, could kill |
| 297 | // a manual pick that started in the meantime. |
| 298 | System.err.println( |
| 299 | "[EarhornCast/RawObserver] follow restore: connect to " + |
| 300 | "'${device.friendlyName}' failed " + |
| 301 | "(${e.javaClass.simpleName}: ${e.message}); keeping the " + |
| 302 | "persisted follow so the next launch retries", |
| 303 | ) |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | /** |
| 308 | * The restore's one dial of [target]'s last known host. Returns true when |
| 309 | * there is nothing left to wait for (it attached, or a follow is already |
| 310 | * running); false to keep waiting for discovery. Same supersede rules as |
| 311 | * the discovery wait. Goes through [observe] (session lock, generation |
| 312 | * check), bounded by [ATTACH_TIMEOUT_MS]. |
| 313 | */ |
| 314 | private suspend fun rescueByLastKnownHost(target: String, rendererAtLaunch: String?): Boolean { |
| 315 | val known = settings.lastKnownHosts.value[target] ?: run { |
| 316 | System.err.println( |
| 317 | "[EarhornCast/RawObserver] '$target' not discovered after " + |
| 318 | "${LAST_KNOWN_HOST_GRACE_MS / 1000}s and no last known host — still waiting", |
| 319 | ) |
| 320 | return false |
| 321 | } |
| 322 | if (settings.followedCastId.value != target || |
| 323 | settings.activeRendererUdn.value != rendererAtLaunch |
| 324 | ) return false // the probe reports the supersede |
| 325 | if (_observing.value) return true // the user beat us to it |
| 326 | val name = known.friendlyName.ifBlank { target } |
| 327 | System.err.println( |
| 328 | "[EarhornCast/RawObserver] '$name' not discovered after " + |
| 329 | "${LAST_KNOWN_HOST_GRACE_MS / 1000}s — trying its last known host", |
| 330 | ) |
| 331 | val device = CastDevice( |
| 332 | id = target, |
| 333 | friendlyName = name, |
| 334 | isGroup = false, |
| 335 | backing = su.litvak.chromecast.api.v2.ChromeCast(known.host, known.port), |
| 336 | ) |
| 337 | // As in the discovery path below: observe() cancels restoreJob, and |
| 338 | // we are running inside it, so drop the handle first. |
| 339 | val self = kotlinx.coroutines.currentCoroutineContext()[Job] |
| 340 | restoreJob = null |
| 341 | val attached = try { |
| 342 | withTimeoutOrNull(ATTACH_TIMEOUT_MS) { observe(device) } ?: false |
| 343 | } catch (e: CancellationException) { |
| 344 | throw e |
| 345 | } catch (e: Exception) { |
| 346 | System.err.println( |
| 347 | "[EarhornCast/RawObserver] last known host for '$name' failed " + |
| 348 | "(${e.javaClass.simpleName}: ${e.message}); keeping the follow, " + |
| 349 | "waiting for discovery", |
| 350 | ) |
| 351 | false |
| 352 | } |
| 353 | if (attached) { |
| 354 | System.err.println("[EarhornCast/RawObserver] attached to '$name' via its last known host") |
| 355 | return true |
| 356 | } |
| 357 | // Still waiting: let a manual pick or stop cancel the wait again. |
| 358 | if (restoreJob == null) restoreJob = self |
| 359 | return _observing.value |
| 360 | } |
| 361 | |
| 362 | /** Tri-state result of one probe of the restore-watcher's combined flow. */ |
| 363 | private sealed interface RestoreProbe { |
| 364 | data object Waiting : RestoreProbe |
| 365 | data object Superseded : RestoreProbe |
| 366 | data class Found(val device: CastDevice) : RestoreProbe |
| 367 | } |
| 368 | |
| 369 | // ── Observe a Cast device ─────────────────────────────────── |
| 370 | |
| 371 | /** |
| 372 | * Begin observing [device]. Tears down any prior observation, opens a fresh |
| 373 | * TLS connection, and starts pushing into [playbackState]. |
| 374 | * |
| 375 | * This class doesn't touch the local renderer: the UI surfaces |
| 376 | * (NowPlayingPane / MPRIS / transport buttons) flip to the observed |
| 377 | * session because `tvState`/`tvTrack` resolve to it while [observing] is |
| 378 | * true. NowPlayingViewModel pauses local playback when following starts |
| 379 | * (ported from Android: otherwise both would be audible with no way to |
| 380 | * reach the local controls). |
| 381 | * |
| 382 | * Returns true when the session attached. Throws when the connection |
| 383 | * fails (after resetting the session; the persisted id stays). Returns |
| 384 | * false if the device has no resolvable host, or if the user stopped |
| 385 | * following while the connection was being made. |
| 386 | * |
| 387 | * Serialized: concurrent calls (startup restore, manual pick, leader |
| 388 | * move) run one after another, never interleaved. |
| 389 | */ |
| 390 | suspend fun observe(device: CastDevice): Boolean = observeOutcome(device) == AttachOutcome.Attached |
| 391 | |
| 392 | private suspend fun observeOutcome(device: CastDevice): AttachOutcome { |
| 393 | // A manual pick supersedes any pending startup restore. (The restore |
| 394 | // path itself nulls restoreJob before calling observe(), so this can |
| 395 | // never cancel the coroutine it's running in.) |
| 396 | restoreJob?.cancel() |
| 397 | restoreJob = null |
| 398 | return sessionMutex.withLock { observeLocked(device) } |
| 399 | } |
| 400 | |
| 401 | /** Session bring-up. Caller holds [sessionMutex]. */ |
| 402 | private suspend fun observeLocked(device: CastDevice): AttachOutcome { |
| 403 | // Recycle any prior session via teardownSession, NOT disconnect(): |
| 404 | // disconnect() clears the persisted follow id, and switching from |
| 405 | // device A to device B must not pass through a persisted-null window |
| 406 | // an ill-timed crash could freeze into "follow nothing next launch." |
| 407 | teardownSession() |
| 408 | val gen = sessionGen.get() |
| 409 | val host = device.host ?: run { |
| 410 | System.err.println( |
| 411 | "[EarhornCast/RawObserver] cannot observe ${device.friendlyName}: no host", |
| 412 | ) |
| 413 | return AttachOutcome.Failed |
| 414 | } |
| 415 | |
| 416 | // Persist the follow target: this is what lets a follow survive an app |
| 417 | // restart (see class KDoc). Recorded before connect() so that intent — |
| 418 | // not connection success — is what's remembered; if the device drops |
| 419 | // the handshake now, the next launch still knows to retry it. |
| 420 | settings.setFollowedCastId(device.id) |
| 421 | if (gen != sessionGen.get()) return AttachOutcome.Superseded |
| 422 | |
| 423 | val c = RawCastClient(host, device.port) |
| 424 | val sc = CoroutineScope(SupervisorJob() + Dispatchers.IO) |
| 425 | client = c |
| 426 | scope = sc |
| 427 | _observing.value = true |
| 428 | _connectedDeviceName.value = device.friendlyName |
| 429 | |
| 430 | try { |
| 431 | c.connect() |
| 432 | } catch (t: Throwable) { |
| 433 | if (gen != sessionGen.get()) { |
| 434 | // disconnect() ran during the connect and already reset |
| 435 | // everything (closing the socket is what made connect throw). |
| 436 | c.disconnect() |
| 437 | sc.cancel() |
| 438 | if (t is CancellationException) throw t |
| 439 | return AttachOutcome.Superseded |
| 440 | } |
| 441 | teardownSession() |
| 442 | throw t |
| 443 | } |
| 444 | if (gen != sessionGen.get()) { |
| 445 | // The user stopped following while we connected. disconnect() may |
| 446 | // have run before client/scope were published, so clean up here. |
| 447 | c.disconnect() |
| 448 | sc.cancel() |
| 449 | if (client === c) teardownSession() |
| 450 | return AttachOutcome.Superseded |
| 451 | } |
| 452 | // Remember where it answered, for the startup rescue when discovery |
| 453 | // is dead (Android parity). On the lifetime scope so the session lock |
| 454 | // isn't held across the disk write, as on Android. |
| 455 | lifecycleScope.launch { |
| 456 | settings.setLastKnownHost(device.id, device.friendlyName, host, device.port) |
| 457 | } |
| 458 | |
| 459 | // Bridge MediaStatus → PlaybackState. Note `current` is null until the |
| 460 | // first MEDIA_STATUS with a non-blank title arrives; the VM bridges |
| 461 | // this gap with a `lastObservedTrack` flow so the UI doesn't blink |
| 462 | // back to a stale local-renderer track between cast broadcasts. |
| 463 | // |
| 464 | // Volume is a special case: we ONLY mirror `ms.volume` into the |
| 465 | // playback-state when the client has confirmed it received an |
| 466 | // authoritative reading from the wire (`volumeKnown = true`). Until |
| 467 | // then, leave the prior value in [_playbackState] alone — the |
| 468 | // PlaybackState() default of 100% would otherwise be presented to |
| 469 | // the UI as if it were the truth, causing the "click anywhere on |
| 470 | // the slider → audible jump from real-5% to clicked-30%" symptom |
| 471 | // that feels like a volume blast. |
| 472 | var lastSeq = -1L |
| 473 | var prevReport: RawCastClient.MediaStatus? = null |
| 474 | sc.launch { |
| 475 | c.mediaStatus.collect { ms -> |
| 476 | // A body already running when teardown cancels this session |
| 477 | // must not write the old device's state over the reset. |
| 478 | if (client !== c) return@collect |
| 479 | // Only a real MEDIA_STATUS (new statusSeq) carries a fresh |
| 480 | // position. A RECEIVER_STATUS volume copy repeats the old one; |
| 481 | // taking it used to snap the bar back to the stale value |
| 482 | // every few seconds (ported from Android N.1, 2026-09-04). |
| 483 | val freshReport = ms.statusSeq != lastSeq |
| 484 | val now = System.currentTimeMillis() |
| 485 | if (freshReport) { |
| 486 | lastSeq = ms.statusSeq |
| 487 | trackFreeze(prevReport, ms, now) |
| 488 | prevReport = ms |
| 489 | if (silenceDeclared) { |
| 490 | silenceDeclared = false |
| 491 | System.err.println("[EarhornCast/RawObserver] statuses resumed — silence declaration withdrawn") |
| 492 | } |
| 493 | } |
| 494 | _playbackState.update { prior -> buildState(ms, prior, freshReport) } |
| 495 | // Freeze latch (Android, 2026-08-30): PLAYING/BUFFERING with a |
| 496 | // pinned position for ZOMBIE_FREEZE_MS = dead session. |
| 497 | if (positionFrozen && !zombieDeclared && frozenSinceMs > 0 && |
| 498 | now - frozenSinceMs > ZOMBIE_FREEZE_MS |
| 499 | ) { |
| 500 | zombieDeclared = true |
| 501 | System.err.println( |
| 502 | "[EarhornCast/RawObserver] ${ms.playerState} with frozen position for " + |
| 503 | "${ZOMBIE_FREEZE_MS / 1000}s — declaring session dead", |
| 504 | ) |
| 505 | movedLeader(device, host)?.let { followMovedLeader(c, it, "frozen session") } |
| 506 | } |
| 507 | // Positive-evidence liveness. IDLE with a lingering session |
| 508 | // (queue finished) is NOT live. Track-boundary IDLE flaps are |
| 509 | // absorbed by the consumer's 20 s debounce. |
| 510 | if (client === c) { |
| 511 | _sessionLive.value = !zombieDeclared && !silenceDeclared && |
| 512 | ms.mediaSessionId != 0 && ms.playerState != "IDLE" |
| 513 | } |
| 514 | } |
| 515 | } |
| 516 | // musicnow is polled only while the session is live (below); a |
| 517 | // session coming (back) to life re-arms the poller at once. |
| 518 | sc.launch { |
| 519 | _sessionLive.collect { live -> if (live) radioWake.trySend(Unit) } |
| 520 | } |
| 521 | |
| 522 | // ── Follower-queue manager ──────────────────────────────── |
| 523 | // Single serialized collector for all wire-level queue traffic. Every |
| 524 | // mutation of [queueOrder]/[queueItemsById] happens here (one logical |
| 525 | // thread, no locking); reads elsewhere see immutable snapshots. |
| 526 | // |
| 527 | // Implementation parameters match the shared cross-platform spec so |
| 528 | // desktop and Android behave identically: batch size 20 for |
| 529 | // QUEUE_GET_ITEMS; publish progressively as batches arrive rather |
| 530 | // than waiting for completeness; debounce QUEUE_CHANGE-triggered |
| 531 | // refetches by 300 ms to coalesce bursts during bulk sender edits; |
| 532 | // on mediaSessionId change (new load = new queue) clear everything |
| 533 | // and refetch; on a refetch, retain buffered items whose ids survive |
| 534 | // so the list doesn't flash empty. |
| 535 | sc.launch { |
| 536 | c.queueEvents.collect { ev -> |
| 537 | when (ev) { |
| 538 | is RawCastClient.QueueEvent.SessionChanged -> { |
| 539 | queueChangeDebounce?.cancel() |
| 540 | queueOrder = emptyList() |
| 541 | queueItemsById = emptyMap() |
| 542 | publishQueue(c) |
| 543 | runCatching { c.requestQueueItemIds() } |
| 544 | } |
| 545 | |
| 546 | is RawCastClient.QueueEvent.ItemIds -> { |
| 547 | val idSet = ev.itemIds.toSet() |
| 548 | queueOrder = ev.itemIds |
| 549 | // Retain survivors so the rows the user is looking at |
| 550 | // don't blank out while the delta refetches. |
| 551 | queueItemsById = queueItemsById.filterKeys { it in idSet } |
| 552 | publishQueue(c) |
| 553 | val missing = ev.itemIds.filter { it !in queueItemsById } |
| 554 | missing.chunked(QUEUE_FETCH_BATCH).forEach { batch -> |
| 555 | runCatching { c.requestQueueItems(batch) } |
| 556 | } |
| 557 | } |
| 558 | |
| 559 | is RawCastClient.QueueEvent.Items -> { |
| 560 | if (ev.items.isNotEmpty()) { |
| 561 | val merged = queueItemsById.toMutableMap() |
| 562 | ev.items.forEach { merged[it.itemId] = it.toTrack() } |
| 563 | queueItemsById = merged |
| 564 | publishQueue(c) |
| 565 | } |
| 566 | } |
| 567 | |
| 568 | is RawCastClient.QueueEvent.Change -> { |
| 569 | // The receiver broadcasts one of these for EVERY sender |
| 570 | // edit; a bulk edit produces a burst. Debounce so the |
| 571 | // burst coalesces into a single id-list round trip. |
| 572 | queueChangeDebounce?.cancel() |
| 573 | queueChangeDebounce = sc.launch { |
| 574 | delay(QUEUE_CHANGE_DEBOUNCE_MS) |
| 575 | runCatching { c.requestQueueItemIds() } |
| 576 | } |
| 577 | } |
| 578 | } |
| 579 | } |
| 580 | } |
| 581 | |
| 582 | // Reconnect supervisor (ported from Android, 2026-09-04). One attempt |
| 583 | // after 5 s and then a blank screen was the old rule; now up to |
| 584 | // RECONNECT_ATTEMPTS, RECONNECT_GAP_MS apart, keeping the last state |
| 585 | // on screen. If the group's leader moved (same name, new host in |
| 586 | // discovery) follow it. If every attempt fails, release with a |
| 587 | // message instead of holding "buffering" forever. |
| 588 | sc.launch { |
| 589 | c.connected.collect { connected -> |
| 590 | if (connected || !_observing.value) return@collect |
| 591 | _playbackState.update { it.copy(isBuffering = true, isPlaying = false) } |
| 592 | var attempt = 0 |
| 593 | while (_observing.value && client === c && attempt < RECONNECT_ATTEMPTS) { |
| 594 | attempt++ |
| 595 | delay(RECONNECT_GAP_MS) |
| 596 | if (!_observing.value || client !== c) return@collect |
| 597 | val moved = movedLeader(device, host) |
| 598 | if (moved != null) { |
| 599 | followMovedLeader(c, moved, "connection lost") |
| 600 | return@collect |
| 601 | } |
| 602 | try { |
| 603 | c.connect() |
| 604 | return@collect |
| 605 | } catch (e: CancellationException) { |
| 606 | throw e |
| 607 | } catch (e: Exception) { |
| 608 | System.err.println( |
| 609 | "[EarhornCast/RawObserver] reconnect attempt $attempt to '${device.friendlyName}' " + |
| 610 | "failed (${e.javaClass.simpleName}: ${e.message})", |
| 611 | ) |
| 612 | } |
| 613 | } |
| 614 | if (_observing.value && client === c) { |
| 615 | System.err.println("[EarhornCast/RawObserver] '${device.friendlyName}' unreachable after $RECONNECT_ATTEMPTS attempts — releasing") |
| 616 | lastFailure.value = device.friendlyName |
| 617 | releaseSession(c) |
| 618 | } |
| 619 | } |
| 620 | } |
| 621 | |
| 622 | // Silence watchdog (Android F2): a session that has sent no playback |
| 623 | // report for STATUS_SILENCE_MS is dead. sessionLive drops (the screen |
| 624 | // shows "Waiting for <device>…" after the consumer's debounce); the |
| 625 | // last state stays in [playbackState] and the next real status |
| 626 | // withdraws the declaration. Polling continues so a recovered leader |
| 627 | // is picked up, and a moved leader is followed. |
| 628 | sc.launch { |
| 629 | var lastSeq = -1L |
| 630 | var silentSince = System.currentTimeMillis() |
| 631 | while (true) { |
| 632 | delay(10_000) |
| 633 | val seq = c.mediaStatus.value.statusSeq |
| 634 | val now = System.currentTimeMillis() |
| 635 | if (seq != lastSeq) { lastSeq = seq; silentSince = now } |
| 636 | else if (!silenceDeclared && now - silentSince > STATUS_SILENCE_MS && |
| 637 | _sessionLive.value && client === c |
| 638 | ) { |
| 639 | System.err.println("[EarhornCast/RawObserver] no playback report for ${STATUS_SILENCE_MS / 1000}s — session dead") |
| 640 | silenceDeclared = true |
| 641 | _sessionLive.value = false |
| 642 | movedLeader(device, host)?.let { followMovedLeader(c, it, "status silence") } |
| 643 | } |
| 644 | } |
| 645 | } |
| 646 | |
| 647 | // Station overlay from musicnow. Android's discipline (0.4.20 |
| 648 | // F-series): a failed read keeps the last good overlay and backs off |
| 649 | // (doubling, capped at 60 s) instead of blanking it; NULL_READS_BEFORE_LATCH |
| 650 | // consecutive station:null reads latch the poller off so a library |
| 651 | // session doesn't poll musicnow forever. A latched poller re-arms when |
| 652 | // the session's content changes, or immediately on |
| 653 | // [requestStatusRefresh] — the in-app "refresh" for the station overlay. |
| 654 | // |
| 655 | // A changed overlay is applied by rebuilding the state directly from |
| 656 | // the client's current MediaStatus. The old path asked the receiver |
| 657 | // for a fresh MEDIA_STATUS and relied on the collector to rebuild, but |
| 658 | // the overlay never reached the screen when no new status arrived. |
| 659 | sc.launch { |
| 660 | c.mediaStatus |
| 661 | .map { it.contentId to it.title } |
| 662 | .distinctUntilChanged() |
| 663 | .drop(1) |
| 664 | .collect { radioWake.trySend(Unit) } |
| 665 | } |
| 666 | sc.launch { |
| 667 | settings.musicnowStatusUrl.drop(1).collect { radioWake.trySend(Unit) } |
| 668 | } |
| 669 | sc.launch { |
| 670 | delay(kotlin.random.Random.nextLong(0, 3_000)) // de-sync the fleet's polls |
| 671 | var nullReads = 0 |
| 672 | var failures = 0 |
| 673 | var latched = false |
| 674 | while (isActive) { |
| 675 | // Android audit 2026-09-02: never poll musicnow for a session |
| 676 | // that isn't live (dead, finished, or not yet reporting). The |
| 677 | // sessionLive collector above sends a wake when it comes back. |
| 678 | if (!_sessionLive.value) { |
| 679 | radioWake.receive() |
| 680 | if (!_sessionLive.value) continue |
| 681 | nullReads = 0; failures = 0; latched = false |
| 682 | delay(kotlin.random.Random.nextLong(0, 3_000)) // keep the fleet de-synced |
| 683 | continue |
| 684 | } |
| 685 | val group = _connectedDeviceName.value ?: device.friendlyName |
| 686 | when (val r = radio.poll(group)) { |
| 687 | is dev.stave.cast.radio.RadioStatusClient.Poll.Station -> { |
| 688 | nullReads = 0; failures = 0 |
| 689 | applyRadio(c, r.now) |
| 690 | } |
| 691 | dev.stave.cast.radio.RadioStatusClient.Poll.NoStation -> { |
| 692 | failures = 0 |
| 693 | if (++nullReads >= dev.stave.cast.radio.RadioStatusClient.NULL_READS_BEFORE_LATCH) { |
| 694 | applyRadio(c, null) |
| 695 | latched = true |
| 696 | } |
| 697 | } |
| 698 | dev.stave.cast.radio.RadioStatusClient.Poll.Failed -> failures++ |
| 699 | dev.stave.cast.radio.RadioStatusClient.Poll.Disabled -> { |
| 700 | applyRadio(c, null) |
| 701 | latched = true |
| 702 | } |
| 703 | } |
| 704 | val woke = if (latched) { |
| 705 | radioWake.receive(); true |
| 706 | } else { |
| 707 | val interval = if (failures == 0) dev.stave.cast.radio.RadioStatusClient.POLL_INTERVAL_MS |
| 708 | else (dev.stave.cast.radio.RadioStatusClient.POLL_INTERVAL_MS shl failures.coerceAtMost(3)) |
| 709 | .coerceAtMost(dev.stave.cast.radio.RadioStatusClient.MAX_BACKOFF_MS) |
| 710 | withTimeoutOrNull(interval) { radioWake.receive() } != null |
| 711 | } |
| 712 | if (woke) { nullReads = 0; failures = 0; latched = false } |
| 713 | } |
| 714 | } |
| 715 | |
| 716 | // Android's rates: media status every 10 s; receiver status only when |
| 717 | // the last one is older than 25 s (was both every 3 s). |
| 718 | sc.launch { |
| 719 | while (true) { |
| 720 | delay(10_000) |
| 721 | runCatching { c.requestMediaStatus() } |
| 722 | if (System.currentTimeMillis() - c.lastReceiverStatusAt > 25_000L) { |
| 723 | runCatching { c.requestReceiverStatus() } |
| 724 | } |
| 725 | } |
| 726 | } |
| 727 | |
| 728 | // Position interpolation — tick progress locally every second while |
| 729 | // playing so the seek bar moves smoothly between MEDIA_STATUS updates |
| 730 | // (which arrive irregularly, every few seconds at best). |
| 731 | // update{} so a status that lands between the read and the write |
| 732 | // isn't overwritten by the old track. Holds while the position is |
| 733 | // frozen or the session is declared dead (no bar running over silence). |
| 734 | sc.launch { |
| 735 | while (true) { |
| 736 | delay(1_000) |
| 737 | if (client !== c) break |
| 738 | if (positionFrozen || zombieDeclared || silenceDeclared) continue |
| 739 | _playbackState.update { s -> |
| 740 | if (s.isPlaying && s.duration > Duration.ZERO) { |
| 741 | s.copy(position = (s.position + 1.seconds).coerceAtMost(s.duration)) |
| 742 | } else s |
| 743 | } |
| 744 | } |
| 745 | } |
| 746 | |
| 747 | // Volume bootstrap. The initial CONNECT + GET_STATUS to NS_RECEIVER |
| 748 | // that c.connect() fires should already include the device's volume |
| 749 | // in its response, but on some devices (and on every CCA group |
| 750 | // leader we've tested) the very first RECEIVER_STATUS response |
| 751 | // comes back without a `volume` sub-object — the receiver only |
| 752 | // includes it on subsequent broadcasts triggered by an actual |
| 753 | // volume change. If we relied on those, an idle session could go |
| 754 | // minutes without ever giving us an authoritative volume reading, |
| 755 | // and the slider would sit at the default 100% the whole time. |
| 756 | // |
| 757 | // Hit NS_RECEIVER GET_STATUS on an exponential-ish backoff until |
| 758 | // volumeKnown latches true (caps at 8 attempts ≈ 12s of wallclock). |
| 759 | // The cost is at most 8 small JSON exchanges; once the latch |
| 760 | // closes, the loop exits and we rely on spontaneous broadcasts + |
| 761 | // the regular periodic refresh for ongoing updates. |
| 762 | sc.launch { |
| 763 | var delay = VOLUME_BOOTSTRAP_INITIAL_DELAY_MS |
| 764 | repeat(VOLUME_BOOTSTRAP_MAX_ATTEMPTS) { |
| 765 | if (c.mediaStatus.value.volumeKnown) return@launch |
| 766 | delay(delay) |
| 767 | if (c.mediaStatus.value.volumeKnown) return@launch |
| 768 | runCatching { c.requestReceiverStatus() } |
| 769 | delay = (delay * 3 / 2).coerceAtMost(VOLUME_BOOTSTRAP_MAX_DELAY_MS) |
| 770 | } |
| 771 | if (!c.mediaStatus.value.volumeKnown) { |
| 772 | System.err.println( |
| 773 | "[EarhornCast/RawObserver] gave up bootstrapping device volume " + |
| 774 | "after $VOLUME_BOOTSTRAP_MAX_ATTEMPTS attempts; opening " + |
| 775 | "the volume input gate so the user can still drive the " + |
| 776 | "slider. First successful SET_VOLUME should cause the " + |
| 777 | "device to broadcast its volume and latch volumeKnown." |
| 778 | ) |
| 779 | _forceVolumeInputAllowed.value = true |
| 780 | } |
| 781 | } |
| 782 | return AttachOutcome.Attached |
| 783 | } |
| 784 | |
| 785 | /** Update the freeze state from two consecutive playback reports (ported |
| 786 | * from Android). PLAYING *or* BUFFERING counts: the 2026-08-31 zombie |
| 787 | * froze in BUFFERING. */ |
| 788 | private fun trackFreeze(prev: RawCastClient.MediaStatus?, ms: RawCastClient.MediaStatus, now: Long) { |
| 789 | val active = setOf("PLAYING", "BUFFERING") |
| 790 | val frozenNow = prev != null && |
| 791 | ms.playerState in active && prev.playerState in active && |
| 792 | ms.mediaSessionId == prev.mediaSessionId && |
| 793 | ms.currentTime > 0.0 && ms.currentTime == prev.currentTime |
| 794 | if (frozenNow == positionFrozen) return |
| 795 | positionFrozen = frozenNow |
| 796 | frozenSinceMs = if (frozenNow) now else 0L |
| 797 | if (!frozenNow && zombieDeclared) { |
| 798 | zombieDeclared = false |
| 799 | System.err.println("[EarhornCast/RawObserver] position moving again — frozen-session declaration withdrawn") |
| 800 | } |
| 801 | } |
| 802 | |
| 803 | /** The same group under a different host in discovery: its leader moved. */ |
| 804 | private fun movedLeader(device: CastDevice, host: String): CastDevice? = |
| 805 | castIntegration.devices.value.firstOrNull { |
| 806 | it.friendlyName == device.friendlyName && it.host != null && it.host != host |
| 807 | } |
| 808 | |
| 809 | /** Follow a moved group leader. Runs on [lifecycleScope], NOT the session |
| 810 | * scope: the new observe tears the current session down first, which |
| 811 | * cancelled the very coroutine that asked for the move, so the follow |
| 812 | * silently ended instead of moving (review P1-4). Skipped if [c]'s |
| 813 | * session has already been replaced. */ |
| 814 | private fun followMovedLeader(c: RawCastClient, moved: CastDevice, reason: String) { |
| 815 | System.err.println("[EarhornCast/RawObserver] leader for '${moved.friendlyName}' moved to ${moved.host} ($reason) — following") |
| 816 | lifecycleScope.launch { |
| 817 | val outcome = try { |
| 818 | sessionMutex.withLock { |
| 819 | if (client !== c) AttachOutcome.Superseded else observeLocked(moved) |
| 820 | } |
| 821 | } catch (e: CancellationException) { |
| 822 | throw e |
| 823 | } catch (e: Exception) { |
| 824 | System.err.println("[EarhornCast/RawObserver] following the moved leader failed (${e.javaClass.simpleName}: ${e.message})") |
| 825 | AttachOutcome.Failed |
| 826 | } |
| 827 | if (outcome == AttachOutcome.Failed) lastFailure.value = moved.friendlyName |
| 828 | } |
| 829 | } |
| 830 | |
| 831 | /** End [c]'s session (under the lock, so it can't hit a newer one). */ |
| 832 | private fun releaseSession(c: RawCastClient) { |
| 833 | lifecycleScope.launch { |
| 834 | sessionMutex.withLock { if (client === c) teardownSession() } |
| 835 | } |
| 836 | } |
| 837 | |
| 838 | |
| 839 | /** Manual pick (design decision 2026-09-04): try [device]; if it cannot |
| 840 | * be reached within [ATTACH_TIMEOUT_MS], put the previous follow back and |
| 841 | * report the name. While it runs, [pendingTargetName] names [device] and |
| 842 | * further picks are ignored (return false). Returns true only when |
| 843 | * [device] attached. */ |
| 844 | suspend fun observeOrRestore(device: CastDevice): Boolean { |
| 845 | if (!_pendingTargetName.compareAndSet(null, device.friendlyName)) { |
| 846 | System.err.println( |
| 847 | "[EarhornCast/RawObserver] pick of '${device.friendlyName}' ignored — " + |
| 848 | "still connecting to '${_pendingTargetName.value}'", |
| 849 | ) |
| 850 | return false |
| 851 | } |
| 852 | val previous = castIntegration.devices.value.firstOrNull { it.friendlyName == _connectedDeviceName.value } |
| 853 | val persistedBefore = settings.followedCastId.value |
| 854 | val outcome = try { |
| 855 | withTimeoutOrNull(ATTACH_TIMEOUT_MS) { observeOutcome(device) } ?: run { |
| 856 | System.err.println("[EarhornCast/RawObserver] '${device.friendlyName}' did not answer within ${ATTACH_TIMEOUT_MS / 1000}s") |
| 857 | AttachOutcome.Failed |
| 858 | } |
| 859 | } catch (e: CancellationException) { |
| 860 | throw e |
| 861 | } catch (e: Exception) { |
| 862 | System.err.println("[EarhornCast/RawObserver] pick of '${device.friendlyName}' failed (${e.javaClass.simpleName}: ${e.message})") |
| 863 | AttachOutcome.Failed |
| 864 | } finally { |
| 865 | _pendingTargetName.compareAndSet(device.friendlyName, null) |
| 866 | } |
| 867 | when (outcome) { |
| 868 | AttachOutcome.Attached -> return true |
| 869 | AttachOutcome.Superseded -> return false // the user stopped following meanwhile |
| 870 | AttachOutcome.Failed -> Unit |
| 871 | } |
| 872 | System.err.println("[EarhornCast/RawObserver] pick of '${device.friendlyName}' failed — restoring ${previous?.friendlyName ?: "nothing"}") |
| 873 | lastFailure.value = device.friendlyName |
| 874 | val alreadyBack = previous != null && _observing.value && _connectedDeviceName.value == previous.friendlyName |
| 875 | if (previous != null && previous.id != device.id && !alreadyBack) { |
| 876 | try { |
| 877 | observeOutcome(previous) |
| 878 | } catch (e: CancellationException) { |
| 879 | throw e |
| 880 | } catch (e: Exception) { |
| 881 | System.err.println("[EarhornCast/RawObserver] restoring '${previous.friendlyName}' failed (${e.javaClass.simpleName}: ${e.message})") |
| 882 | } |
| 883 | } else if (previous == null && settings.followedCastId.value == device.id) { |
| 884 | // Nothing to go back to: don't auto-follow, at the next launch, a |
| 885 | // device the user just couldn't reach. Put back what was persisted. |
| 886 | settings.setFollowedCastId(persistedBefore) |
| 887 | } |
| 888 | return false |
| 889 | } |
| 890 | |
| 891 | /** |
| 892 | * Stop observing AND forget the persisted follow target. Idempotent — safe |
| 893 | * to call when not observing. |
| 894 | * |
| 895 | * This is the user-facing stop (the "Stop following" row and the eyeball |
| 896 | * toggle are its only callers): an explicit stop means "don't resume this |
| 897 | * on the next launch either," so the persisted id is cleared. Internal |
| 898 | * paths that must keep the persisted id — session recycling at the top of |
| 899 | * [observe], failed-restore cleanup — use [teardownSession] directly. |
| 900 | * Nothing on the app-shutdown path calls this (deliberately: the process |
| 901 | * dying while following is precisely the case the restore exists for). |
| 902 | */ |
| 903 | override fun disconnect() { |
| 904 | restoreJob?.cancel() |
| 905 | restoreJob = null |
| 906 | teardownSession() |
| 907 | // Non-suspend interface method, so the write hops onto the lifetime |
| 908 | // scope — NOT the per-session scope, which teardownSession just killed. |
| 909 | lifecycleScope.launch { settings.setFollowedCastId(null) } |
| 910 | } |
| 911 | |
| 912 | /** Tear down the observation session (socket, collectors, state flows) |
| 913 | * without touching the persisted follow id. Idempotent. */ |
| 914 | private fun teardownSession() { |
| 915 | sessionGen.incrementAndGet() |
| 916 | scope?.cancel() |
| 917 | scope = null |
| 918 | client?.disconnect() |
| 919 | client = null |
| 920 | _observing.value = false |
| 921 | _connectedDeviceName.value = null |
| 922 | _sessionLive.value = false |
| 923 | positionFrozen = false |
| 924 | frozenSinceMs = 0L |
| 925 | zombieDeclared = false |
| 926 | silenceDeclared = false |
| 927 | _playbackState.value = PlaybackState() |
| 928 | // Follower-queue state belongs to the session we just left. |
| 929 | queueChangeDebounce = null |
| 930 | queueOrder = emptyList() |
| 931 | queueItemsById = emptyMap() |
| 932 | // Reset the fail-safe — the next device we attach to gets its own |
| 933 | // fresh bootstrap window. Without this, observing device A (which |
| 934 | // timed out) followed by device B would skip the gate for B even |
| 935 | // though we have no authoritative volume from B yet. |
| 936 | _forceVolumeInputAllowed.value = false |
| 937 | radioNow = null |
| 938 | } |
| 939 | |
| 940 | // ── Transport commands (forwarded to the underlying client) ── |
| 941 | // |
| 942 | // Each command optimistically updates [_playbackState] before the wire |
| 943 | // round-trip — the next MEDIA_STATUS will reconfirm or correct as needed. |
| 944 | // Without this, rapid play/pause taps would read the LAST-confirmed state |
| 945 | // (which lags by 150-500ms behind the user's intent) and dispatch the |
| 946 | // wrong command, e.g. tap-Pause → state still says PLAYING → tap-Pause |
| 947 | // again → sends a second PAUSE instead of resuming. Updating eagerly lets |
| 948 | // the toggle logic in NowPlayingViewModel/MprisIntegration compute against |
| 949 | // the user's most recent intent on every press. |
| 950 | |
| 951 | // update{} throughout: these run on the caller's thread while the |
| 952 | // MEDIA_STATUS collector writes the same flow; a plain read-modify-write |
| 953 | // could put the previous track back for a poll cycle. |
| 954 | |
| 955 | override suspend fun play() { |
| 956 | _playbackState.update { it.copy(isPlaying = true, isStopped = false, isBuffering = false) } |
| 957 | client?.play() |
| 958 | } |
| 959 | |
| 960 | override suspend fun pause() { |
| 961 | _playbackState.update { it.copy(isPlaying = false, isBuffering = false) } |
| 962 | client?.pause() |
| 963 | } |
| 964 | |
| 965 | override suspend fun stop() { |
| 966 | _playbackState.update { it.copy(isPlaying = false, isStopped = true, isBuffering = false) } |
| 967 | client?.stop() |
| 968 | } |
| 969 | |
| 970 | override suspend fun next() { |
| 971 | client?.next() |
| 972 | } |
| 973 | |
| 974 | override suspend fun previous() { |
| 975 | client?.previous() |
| 976 | } |
| 977 | |
| 978 | override suspend fun seek(position: Duration) { |
| 979 | // A session with no duration is a live stream (radio): there is |
| 980 | // nothing to seek, and the receiver ignores it (Android: duration is |
| 981 | // the only reliable LIVE discriminator). |
| 982 | if (_playbackState.value.duration <= Duration.ZERO) { |
| 983 | System.err.println("[EarhornCast/RawObserver] seek refused: live stream") |
| 984 | return |
| 985 | } |
| 986 | _playbackState.update { it.copy(position = position) } |
| 987 | client?.seek(position.inWholeMilliseconds / 1000.0) |
| 988 | } |
| 989 | |
| 990 | /** Repeat/shuffle for the OBSERVED session. Cast has one axis; shuffle |
| 991 | * is expressed as REPEAT_ALL_AND_SHUFFLE. */ |
| 992 | override suspend fun setRepeatMode(mode: dev.stave.domain.renderer.RepeatMode) { |
| 993 | client?.setRepeatMode(when (mode) { |
| 994 | dev.stave.domain.renderer.RepeatMode.OFF -> "REPEAT_OFF" |
| 995 | dev.stave.domain.renderer.RepeatMode.ALL -> "REPEAT_ALL" |
| 996 | dev.stave.domain.renderer.RepeatMode.ONE -> "REPEAT_SINGLE" |
| 997 | }) |
| 998 | } |
| 999 | override suspend fun setShuffle(on: Boolean) { |
| 1000 | client?.setRepeatMode(if (on) "REPEAT_ALL_AND_SHUFFLE" else "REPEAT_OFF") |
| 1001 | } |
| 1002 | |
| 1003 | override suspend fun setVolume(percent: Int) { |
| 1004 | val c = client ?: return |
| 1005 | // Refuse to send SET_VOLUME until we've heard an authoritative |
| 1006 | // volume reading from the device — OR the bootstrap timed out and |
| 1007 | // we've force-opened the gate to keep the user unblocked. |
| 1008 | // Otherwise a user click on the (visually unsynced) slider would |
| 1009 | // push e.g. SET_VOLUME{0.30} to a device actually at 5% — a 6x |
| 1010 | // jump that's audibly indistinguishable from a "blast." Once the |
| 1011 | // bootstrap loop in observe() latches volumeKnown, this gate |
| 1012 | // opens and the user's input is honored normally. |
| 1013 | if (!c.mediaStatus.value.volumeKnown && !_forceVolumeInputAllowed.value) { |
| 1014 | System.err.println( |
| 1015 | "[EarhornCast/RawObserver] setVolume($percent) ignored: device " + |
| 1016 | "volume not yet known. Slider input will be honored once the " + |
| 1017 | "device reports its current level." |
| 1018 | ) |
| 1019 | return |
| 1020 | } |
| 1021 | val clamped = percent.coerceIn(0, 100) |
| 1022 | _playbackState.update { it.copy(volumePercent = clamped) } |
| 1023 | c.setVolume(clamped / 100.0) |
| 1024 | } |
| 1025 | |
| 1026 | /** |
| 1027 | * Jump the observed session to the queue row at [index]. Maps the index |
| 1028 | * back to the receiver's itemId (the id order is the source of truth for |
| 1029 | * position) and sends a position-only QUEUE_UPDATE. Optimistically moves |
| 1030 | * the local highlight; the next MEDIA_STATUS's currentItemId reconfirms |
| 1031 | * or corrects. Display-and-jump is the follower's entire queue contract — |
| 1032 | * see [CastObserverApi.jumpToQueueItem] for why editing is prohibited. |
| 1033 | */ |
| 1034 | override suspend fun jumpToQueueItem(index: Int) { |
| 1035 | val c = client ?: return |
| 1036 | val itemId = queueOrder.getOrNull(index) ?: return |
| 1037 | _playbackState.update { it.copy(queueIndex = index) } |
| 1038 | c.queueJumpTo(itemId) |
| 1039 | } |
| 1040 | |
| 1041 | /** Apply a new station overlay (or clear it) and republish immediately. */ |
| 1042 | private fun applyRadio(c: RawCastClient, now: dev.stave.cast.radio.RadioStatusClient.RadioNow?) { |
| 1043 | if (now == radioNow) return |
| 1044 | radioNow = now |
| 1045 | if (client !== c) return |
| 1046 | // Swap only the displayed track. Transport fields stay as they are, so |
| 1047 | // an optimistic play/pause the user just pressed isn't rolled back to |
| 1048 | // the last receiver report. update{} keeps this atomic against the |
| 1049 | // MEDIA_STATUS collector, which writes the same flow. |
| 1050 | _playbackState.update { prior -> |
| 1051 | prior.copy(current = buildState(c.mediaStatus.value, prior, freshReport = false).current) |
| 1052 | } |
| 1053 | } |
| 1054 | |
| 1055 | /** Map one [RawCastClient.MediaStatus] (plus the current station overlay) |
| 1056 | * to the domain [PlaybackState]. [freshReport] is false for a rebuild that |
| 1057 | * isn't driven by a new MEDIA_STATUS, so the interpolated position is kept. */ |
| 1058 | private fun buildState(ms: RawCastClient.MediaStatus, prior: PlaybackState, freshReport: Boolean): PlaybackState { |
| 1059 | val rn = radioNow |
| 1060 | // The overlay applies ONLY to sessions that ARE radio (no duration), |
| 1061 | // as on Android: a stale overlay, kept through musicnow failures by |
| 1062 | // design, otherwise dressed library playback on the same group in |
| 1063 | // the station's song (Android, 2026-08-29). A station that loads with |
| 1064 | // no title of its own still shows the overlay. |
| 1065 | val overlay = rn?.title != null && ms.duration <= 0.0 && ms.mediaSessionId != 0 |
| 1066 | // Current track: prefer the fetched queue item for the receiver's |
| 1067 | // currentItemId (ported from Android). currentItemId rides on every |
| 1068 | // status push, while the media block is often missing or half-filled |
| 1069 | // during auto-advance, and the client keeps title/artist sticky — so |
| 1070 | // the MEDIA_STATUS path left the previous title up for a track or |
| 1071 | // more. Queue items carry the full per-item metadata, art included. |
| 1072 | val fromQueue = if (overlay) null else ms.currentItemId?.let { queueItemsById[it] } |
| 1073 | return PlaybackState( |
| 1074 | current = if (fromQueue != null) fromQueue.copy( |
| 1075 | // Same identity as the MEDIA_STATUS path (the content id), so |
| 1076 | // enrichment/lyrics keys don't depend on which path built it. |
| 1077 | id = fromQueue.resourceUrl.ifBlank { fromQueue.id }, |
| 1078 | duration = fromQueue.duration ?: ms.duration.seconds, |
| 1079 | ) else if (overlay || !ms.title.isNullOrBlank()) DidlItem.Track( |
| 1080 | id = if (overlay) "station:${rn!!.station}|${rn.title}|${rn.artist ?: ""}" else ms.contentId ?: "", |
| 1081 | parentId = "", |
| 1082 | title = if (overlay) rn!!.title!! else ms.title ?: "", |
| 1083 | artist = if (overlay) rn!!.artist else ms.artist, |
| 1084 | albumArtist = if (overlay) rn!!.artist else ms.artist, |
| 1085 | album = if (overlay) rn!!.station else ms.album, |
| 1086 | composer = null, |
| 1087 | genre = null, |
| 1088 | date = null, |
| 1089 | trackNumber = null, |
| 1090 | discNumber = null, |
| 1091 | duration = ms.duration.seconds, |
| 1092 | albumArtUri = gateArt(if (overlay) (rn!!.albumArtUrl ?: ms.albumArtUrl) else ms.albumArtUrl), |
| 1093 | resourceUrl = if (overlay) "station:${rn!!.station}|${rn.title}|${rn.artist ?: ""}" else ms.contentId ?: "", |
| 1094 | mimeType = null, |
| 1095 | sampleFrequency = null, |
| 1096 | bitsPerSample = null, |
| 1097 | channels = null, |
| 1098 | bitrate = null, |
| 1099 | size = null, |
| 1100 | ) else null, |
| 1101 | // Assembled follower queue (see the queue-event collector |
| 1102 | // below). The index is derived from the receiver's |
| 1103 | // authoritative currentItemId cross-referenced against the |
| 1104 | // fetched id order — NEVER counted locally, so it cannot |
| 1105 | // drift from what's audible. |
| 1106 | queue = buildQueueTracks(), |
| 1107 | queueIndex = ms.currentItemId?.let { queueOrder.indexOf(it) } ?: -1, |
| 1108 | isPlaying = ms.playerState == "PLAYING", |
| 1109 | isBuffering = ms.playerState == "BUFFERING", |
| 1110 | isStopped = ms.playerState == "IDLE", |
| 1111 | position = if (freshReport) ms.currentTime.seconds else prior.position, |
| 1112 | duration = ms.duration.seconds, |
| 1113 | repeatMode = when (ms.repeatMode) { |
| 1114 | "REPEAT_ALL", "REPEAT_ALL_AND_SHUFFLE" -> dev.stave.domain.renderer.RepeatMode.ALL |
| 1115 | "REPEAT_SINGLE" -> dev.stave.domain.renderer.RepeatMode.ONE |
| 1116 | else -> dev.stave.domain.renderer.RepeatMode.OFF |
| 1117 | }, |
| 1118 | shuffleEnabled = ms.repeatMode == "REPEAT_ALL_AND_SHUFFLE", |
| 1119 | volumePercent = if (ms.volumeKnown) |
| 1120 | (ms.volume * 100).toInt() |
| 1121 | else |
| 1122 | prior.volumePercent, |
| 1123 | ) |
| 1124 | } |
| 1125 | |
| 1126 | /** Immutable snapshot of the follower queue as display tracks. Ids whose |
| 1127 | * metadata hasn't arrived yet render as placeholders (progressive |
| 1128 | * publishing) so row positions stay stable while batches land. */ |
| 1129 | private fun buildQueueTracks(): List<DidlItem.Track> { |
| 1130 | val order = queueOrder |
| 1131 | if (order.isEmpty()) return emptyList() |
| 1132 | val items = queueItemsById |
| 1133 | return order.map { id -> items[id] ?: placeholderTrack(id) } |
| 1134 | } |
| 1135 | |
| 1136 | /** Republish the queue, and the current track with it (it may come from |
| 1137 | * a queue item that just arrived). Transport fields are left alone so an |
| 1138 | * optimistic play/pause isn't rolled back. */ |
| 1139 | private fun publishQueue(c: RawCastClient) { |
| 1140 | if (client !== c) return |
| 1141 | _playbackState.update { prior -> |
| 1142 | val built = buildState(c.mediaStatus.value, prior, freshReport = false) |
| 1143 | prior.copy(current = built.current, queue = built.queue, queueIndex = built.queueIndex) |
| 1144 | } |
| 1145 | } |
| 1146 | |
| 1147 | /** Followed-session art comes from whatever app is casting (often a |
| 1148 | * Google/Spotify CDN). Pass it through only when it is on the LAN or the |
| 1149 | * user allows external calls (review 4 #5). */ |
| 1150 | private fun gateArt(url: String?): String? = |
| 1151 | url?.takeIf { dev.stave.net.UrlPolicy.isLanUrl(it) || settings.allowExternalCalls.value } |
| 1152 | |
| 1153 | /** Map one wire-level queue item to the domain Track the UI renders. |
| 1154 | * The id is keyed on the receiver's itemId — unique per queue slot, so |
| 1155 | * the same track queued twice yields two distinct row keys. */ |
| 1156 | private fun RawCastClient.QueueItemLite.toTrack(): DidlItem.Track = |
| 1157 | DidlItem.Track( |
| 1158 | id = "cast-queue:$itemId", |
| 1159 | parentId = "", |
| 1160 | title = title ?: contentId?.substringAfterLast('/') ?: "Unknown track", |
| 1161 | artist = artist, |
| 1162 | albumArtist = artist, |
| 1163 | album = album, |
| 1164 | composer = null, |
| 1165 | genre = null, |
| 1166 | date = null, |
| 1167 | trackNumber = null, |
| 1168 | discNumber = null, |
| 1169 | duration = durationSeconds?.seconds, |
| 1170 | albumArtUri = gateArt(albumArtUrl), |
| 1171 | resourceUrl = contentId ?: "", |
| 1172 | mimeType = null, |
| 1173 | sampleFrequency = null, |
| 1174 | bitsPerSample = null, |
| 1175 | channels = null, |
| 1176 | bitrate = null, |
| 1177 | size = null, |
| 1178 | ) |
| 1179 | |
| 1180 | private fun placeholderTrack(itemId: Int): DidlItem.Track = |
| 1181 | DidlItem.Track( |
| 1182 | id = "cast-queue:$itemId", |
| 1183 | parentId = "", |
| 1184 | title = "…", |
| 1185 | artist = null, |
| 1186 | albumArtist = null, |
| 1187 | album = null, |
| 1188 | composer = null, |
| 1189 | genre = null, |
| 1190 | date = null, |
| 1191 | trackNumber = null, |
| 1192 | discNumber = null, |
| 1193 | duration = null, |
| 1194 | albumArtUri = null, |
| 1195 | resourceUrl = "", |
| 1196 | mimeType = null, |
| 1197 | sampleFrequency = null, |
| 1198 | bitsPerSample = null, |
| 1199 | channels = null, |
| 1200 | bitrate = null, |
| 1201 | size = null, |
| 1202 | ) |
| 1203 | |
| 1204 | /** Force a MEDIA_STATUS refresh. The periodic timer already does this |
| 1205 | * while observing; this is the manual trigger for surfaces that want |
| 1206 | * fresh data immediately (e.g. after the user re-focuses the window). |
| 1207 | * Also re-reads musicnow's /status now, re-arming a latched poller. */ |
| 1208 | override suspend fun requestStatusRefresh() { |
| 1209 | radioWake.trySend(Unit) |
| 1210 | client?.requestMediaStatus() |
| 1211 | } |
| 1212 | |
| 1213 | companion object { |
| 1214 | /** Starting delay before the first volume-bootstrap retry. The |
| 1215 | * initial CONNECT + GET_STATUS that c.connect() fires usually |
| 1216 | * delivers volume on its own response, so the first retry only |
| 1217 | * needs to fire after enough time for that to complete. */ |
| 1218 | private const val VOLUME_BOOTSTRAP_INITIAL_DELAY_MS = 300L |
| 1219 | |
| 1220 | /** Cap on the exponential-ish backoff between retries. Bounds the |
| 1221 | * worst-case wait between calls so a slow / temporarily-laggy |
| 1222 | * device still gets re-asked within seconds. */ |
| 1223 | private const val VOLUME_BOOTSTRAP_MAX_DELAY_MS = 2_500L |
| 1224 | |
| 1225 | /** Maximum number of RECEIVER_STATUS requests before giving up. |
| 1226 | * With the 300→2500ms backoff, 8 attempts ≈ 12s of wallclock — |
| 1227 | * past that, the device clearly isn't going to volunteer a |
| 1228 | * volume reading and the user can still drive the slider; we |
| 1229 | * just won't have the initial sync. */ |
| 1230 | private const val VOLUME_BOOTSTRAP_MAX_ATTEMPTS = 8 |
| 1231 | |
| 1232 | /** QUEUE_GET_ITEMS batch size — shared cross-platform value so |
| 1233 | * desktop and Android generate identical wire traffic patterns. */ |
| 1234 | private const val QUEUE_FETCH_BATCH = 20 |
| 1235 | |
| 1236 | /** Debounce window for QUEUE_CHANGE-triggered id-list refetches. |
| 1237 | * Coalesces the broadcast burst a driving sender produces during a |
| 1238 | * bulk edit into a single round trip. Shared cross-platform value. */ |
| 1239 | private const val QUEUE_CHANGE_DEBOUNCE_MS = 300L |
| 1240 | private const val RECONNECT_ATTEMPTS = 4 |
| 1241 | private const val RECONNECT_GAP_MS = 5_000L |
| 1242 | /** Android F2: no playback report for this long = dead session. */ |
| 1243 | private const val STATUS_SILENCE_MS = 150_000L |
| 1244 | /** PLAYING/BUFFERING with a pinned position this long = dead session |
| 1245 | * (Android's freeze latch; far above musicnow's 90 s skip watchdog |
| 1246 | * by design, so the source-side skip wins when it exists). */ |
| 1247 | const val ZOMBIE_FREEZE_MS = 300_000L |
| 1248 | /** How long a manual follow pick may take before it counts as failed |
| 1249 | * (Android's attach window). */ |
| 1250 | const val ATTACH_TIMEOUT_MS = 20_000L |
| 1251 | /** How long the startup restore waits for discovery before dialling the |
| 1252 | * followed device's last known host (Android's 20 s grace). */ |
| 1253 | const val LAST_KNOWN_HOST_GRACE_MS = 20_000L |
| 1254 | } |
| 1255 | } |