RawCastObserver.kt
Each line, with the release that last changed it.
| 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 | } |