package dev.stave.cast.raw
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import org.json.JSONObject
import java.io.*
import java.net.InetSocketAddress
import javax.net.ssl.SSLContext
import javax.net.ssl.SSLSocket
import javax.net.ssl.TrustManager
import javax.net.ssl.X509TrustManager
/**
* Raw Cast protocol client for observing Chromecast sessions without the Cast SDK.
*
* Connects via TLS to port 8009 on any Chromecast/Chromecast Audio/group leader,
* discovers the active media session, and streams [MediaStatus] updates. Transport
* commands (play/pause/stop/seek/next/previous) are sent as JSON over the media
* namespace.
*
* This bypasses Google Play Services entirely — works on Shield TV, Fire TV, libre
* builds, or any device that can open a TLS socket.
*
* Protocol references:
* - github.com/thibauts/node-castv2
* - github.com/home-assistant-libs/pychromecast
* - chromium.googlesource.com/.../cast_channel.proto
*/
class RawCastClient(
val host: String,
private val port: Int = 8009,
) {
companion object {
// Cast protocol namespaces
private const val NS_CONNECTION = "urn:x-cast:com.google.cast.tp.connection"
private const val NS_HEARTBEAT = "urn:x-cast:com.google.cast.tp.heartbeat"
private const val NS_RECEIVER = "urn:x-cast:com.google.cast.receiver"
private const val NS_MEDIA = "urn:x-cast:com.google.cast.media"
private const val SENDER_ID = "sender-earhorn"
private const val RECEIVER_ID = "receiver-0"
private const val HEARTBEAT_INTERVAL_MS = 5_000L
private const val CONNECT_TIMEOUT_MS = 10_000
}
data class MediaStatus(
val title: String? = null,
val artist: String? = null,
val album: String? = null,
val albumArtUrl: String? = null,
val contentId: String? = null,
val duration: Double = 0.0,
val currentTime: Double = 0.0,
val playerState: String = "UNKNOWN", // PLAYING, PAUSED, IDLE, BUFFERING
val mediaSessionId: Int = 0,
/** Receiver's supportedMediaCommands bitmask (Cast constants:
* 1=PAUSE, 2=SEEK, 64=QUEUE_NEXT, 128=QUEUE_PREV, 256=SHUFFLE,
* 1024|2048=REPEAT). 0 = not yet reported. */
val supportedMediaCommands: Long = 0L,
/** BUFFERED or LIVE; "" until reported. */
val streamType: String = "",
/** The receiver's queue repeat mode: REPEAT_OFF, REPEAT_ALL,
* REPEAT_SINGLE, REPEAT_ALL_AND_SHUFFLE; "" until reported. Read so
* the Shuffle/Repeat controls can show and change the HOUSE's mode
* while observing (I.15, design decision 2026-09-02). */
val repeatMode: String = "",
val volume: Double = 1.0,
val isMuted: Boolean = false,
/** The receiver-side queue item ID of the currently playing item.
* Cross-referenced against [queueItems] to derive the queue index. */
val currentItemId: Int = 0,
/** Advances by one for every MEDIA_STATUS message the receiver sends,
* and for nothing else. RECEIVER_STATUS (device volume) copies this
* object with the same value. The observer keys "is this a new
* playback report?" on this number, not on object identity — a
* volume report used to count as a new status with an unchanged
* position and trip the freeze detector (review N.1, 2026-09-02). */
val statusSeq: Long = 0L,
)
/** One entry of the receiver-side queue, fetched via QUEUE_GET_ITEM_IDS /
* QUEUE_GET_ITEMS. This is the same mechanism Google's own MediaQueue
* class uses to populate sender-side queue views — the receiver is the
* source of truth and any connected sender/observer may read it. */
data class QueueItem(
val itemId: Int,
val contentId: String?,
val title: String?,
val artist: String?,
val album: String?,
val albumArtUrl: String?,
val durationSeconds: Double,
)
private val _mediaStatus = MutableStateFlow(MediaStatus())
val mediaStatus: StateFlow<MediaStatus> = _mediaStatus.asStateFlow()
/** The receiver's queue in playback order. Populated after connect once
* a media session is discovered; refreshed automatically on QUEUE_CHANGE
* broadcasts (fired when the sender edits the queue) and on media-session
* change (new load). Emits progressively as item batches arrive. */
private val _queueItems = MutableStateFlow<List<QueueItem>>(emptyList())
val queueItems: StateFlow<List<QueueItem>> = _queueItems.asStateFlow()
// Queue-fetch bookkeeping. Guarded by [queueLock]: the read loop
// mutates from its handlers, but disconnect()/clearQueueState() can
// arrive from the UI thread mid-iteration — an unsynchronized HashMap
// there is a ConcurrentModificationException crash.
private val queueLock = Any()
private var queueItemIdOrder: List<Int> = emptyList()
private val queueItemBuf = mutableMapOf<Int, QueueItem>()
private var queueRefetchJob: Job? = null
// Last-received timestamps — lets observers gate their periodic pulls
// on staleness instead of hammering the receiver on a fixed tick.
@Volatile private var lastMediaStatusAt = 0L
@Volatile private var lastReceiverStatusAt = 0L
fun mediaStatusAgeMs(): Long = System.currentTimeMillis() - lastMediaStatusAt
fun receiverStatusAgeMs(): Long = System.currentTimeMillis() - lastReceiverStatusAt
private val _connected = MutableStateFlow(false)
val connected: StateFlow<Boolean> = _connected.asStateFlow()
private var socket: SSLSocket? = null
private var output: OutputStream? = null
private var input: InputStream? = null
private var scope: CoroutineScope? = null
private var transportId: String? = null
private var requestId = 1
// ── Connect / Disconnect ─────────────────────────────────────
suspend fun connect() = withContext(Dispatchers.IO) {
disconnect()
val sc = CoroutineScope(SupervisorJob() + Dispatchers.IO)
scope = sc
android.util.Log.i("EarhornCast", "connecting to $host:$port")
// TLS with permissive trust (Chromecast uses self-signed certs)
val sslContext = SSLContext.getInstance("TLS")
sslContext.init(null, arrayOf<TrustManager>(PermissiveTrustManager), null)
val factory = sslContext.socketFactory
val sock = factory.createSocket() as SSLSocket
sock.connect(InetSocketAddress(host, port), CONNECT_TIMEOUT_MS)
// The handshake had no timeout: a receiver that accepts and stalls
// pinned this thread until it let go (I.13). Bound it; a real
// handshake on the LAN completes in well under a second. Reset to
// no timeout afterwards — the read loop blocks on purpose.
sock.soTimeout = CONNECT_TIMEOUT_MS
sock.startHandshake()
sock.soTimeout = 0
socket = sock
output = BufferedOutputStream(sock.outputStream)
input = BufferedInputStream(sock.inputStream)
_connected.value = true
android.util.Log.i("EarhornCast", "TLS handshake OK — sending CONNECT + GET_STATUS")
// Start reader + heartbeat
sc.launch { readLoop() }
sc.launch { heartbeatLoop() }
// Connect to receiver platform
sendJson(RECEIVER_ID, NS_CONNECTION, """{"type":"CONNECT","origin":{}}""")
// Get receiver status to discover active app/session
sendJson(RECEIVER_ID, NS_RECEIVER,
"""{"type":"GET_STATUS","requestId":${nextRequestId()}}""")
}
fun disconnect() {
if (_connected.value || socket != null) {
android.util.Log.i("EarhornCast", "disconnecting from $host:$port")
}
scope?.cancel()
scope = null
runCatching { socket?.close() }
socket = null
output = null
input = null
transportId = null
_connected.value = false
clearQueueState()
}
// ── Transport commands ───────────────────────────────────────
suspend fun play() = sendMediaCommand("PLAY")
suspend fun pause() = sendMediaCommand("PAUSE")
suspend fun stop() = sendMediaCommand("STOP")
suspend fun seek(positionSeconds: Double) {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"SEEK","mediaSessionId":$msid,"currentTime":$positionSeconds,"requestId":${nextRequestId()}}""")
}
suspend fun next() {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_NEXT","mediaSessionId":$msid,"requestId":${nextRequestId()}}""")
}
suspend fun previous() {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_PREV","mediaSessionId":$msid,"requestId":${nextRequestId()}}""")
}
suspend fun setVolume(level: Double) {
sendJson(RECEIVER_ID, NS_RECEIVER,
"""{"type":"SET_VOLUME","volume":{"level":$level},"requestId":${nextRequestId()}}""")
// Optimistic echo: reflect the target immediately so the local UI
// (and any localVol re-sync) never reads a stale level mid-gesture.
// The receiver's RECEIVER_STATUS broadcast confirms shortly after
// and corrects if the device clamped the value.
_mediaStatus.value = _mediaStatus.value.copy(volume = level.coerceIn(0.0, 1.0))
}
/** Re-request receiver status (device volume, running app). The receiver
* broadcasts RECEIVER_STATUS on every volume change from any sender or
* hardware control, but a periodic pull guards against missed
* broadcasts so multi-device observers stay in sync. */
suspend fun requestReceiverStatus() {
sendJson(RECEIVER_ID, NS_RECEIVER,
"""{"type":"GET_STATUS","requestId":${nextRequestId()}}""")
}
/** Re-request current media status from the Chromecast. Called periodically
* to catch any MEDIA_STATUS broadcasts that were missed. */
suspend fun requestMediaStatus() {
val tid = transportId ?: return
sendJson(tid, NS_MEDIA,
"""{"type":"GET_STATUS","requestId":${nextRequestId()}}""")
}
// ── Queue introspection ──────────────────────────────────────
// Two-step fetch, mirroring the Cast SDK's MediaQueue: first ask for
// the ordered list of item IDs, then fetch item metadata in batches.
/** Ask the receiver for the ordered queue item IDs. The response
* arrives as a QUEUE_ITEM_IDS message handled in [handleMessage],
* which then fires batched QUEUE_GET_ITEMS requests. */
suspend fun requestQueueItemIds() {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_GET_ITEM_IDS","mediaSessionId":$msid,"requestId":${nextRequestId()}}""")
}
private suspend fun requestQueueItemsBatch(ids: List<Int>) {
if (ids.isEmpty()) return
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_GET_ITEMS","mediaSessionId":$msid,"itemIds":${ids.joinToString(",", "[", "]")},"requestId":${nextRequestId()}}""")
}
/** Jump playback to a specific queue item. QUEUE_UPDATE with a
* currentItemId changes position without altering queue contents,
* so the sender's local queue copy stays in sync (its index tracker
* follows via onQueueStatusUpdated). */
/** Set the receiver queue's repeat mode (Cast QUEUE_UPDATE.repeatMode). */
suspend fun setRepeatMode(mode: String) {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_UPDATE","mediaSessionId":$msid,"repeatMode":"$mode","requestId":${nextRequestId()}}""")
}
/** Shuffle the receiver queue (Cast QUEUE_UPDATE.shuffle). */
suspend fun setShuffle(on: Boolean) {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_UPDATE","mediaSessionId":$msid,"shuffle":$on,"requestId":${nextRequestId()}}""")
}
suspend fun queueJump(itemId: Int) {
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"QUEUE_UPDATE","mediaSessionId":$msid,"currentItemId":$itemId,"requestId":${nextRequestId()}}""")
}
private suspend fun sendMediaCommand(type: String) {
// Every mutating command is logged WITH ITS CALL SITE. If any screen
// ever kills a session, this line is the proof — and its absence at
// the moment a cast dies is proof Earhorn did not send the kill.
android.util.Log.i("EarhornCast",
"OUTBOUND MEDIA COMMAND: $type to $host", Throwable("call site"))
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
sendJson(tid, NS_MEDIA,
"""{"type":"$type","mediaSessionId":$msid,"requestId":${nextRequestId()}}""")
}
// ── Message I/O ─────────────────────────────────────────────
private suspend fun sendJson(
destinationId: String,
namespace: String,
json: String,
) = withContext(Dispatchers.IO) {
val msg = CastMessageCodec.encode(
sourceId = SENDER_ID,
destinationId = destinationId,
namespace = namespace,
payloadUtf8 = json,
)
val lenBuf = ByteArray(4)
lenBuf[0] = (msg.size shr 24 and 0xFF).toByte()
lenBuf[1] = (msg.size shr 16 and 0xFF).toByte()
lenBuf[2] = (msg.size shr 8 and 0xFF).toByte()
lenBuf[3] = (msg.size and 0xFF).toByte()
val out = output ?: return@withContext
// Every public command (play/pause/volume/queue/…) funnels through
// here, and callers launch them from viewModelScope / UI scopes.
// A write on a dead socket (CCA rebooted, network drop) used to
// throw IOException out of this suspend fun → unhandled coroutine
// exception → APP CRASH, landing the user back on the Browse
// start-destination. Commands are fire-and-forget over an
// unreliable transport: swallow the failure, flip connected=false
// so the observer's supervisor can start reconnecting.
try {
synchronized(out) {
out.write(lenBuf)
out.write(msg)
out.flush()
}
} catch (e: java.io.IOException) {
android.util.Log.w("EarhornCast",
"send failed (${e.message}) — marking connection lost")
_connected.value = false
}
}
private suspend fun readLoop() {
try {
val inp = input ?: return
val lenBuf = ByteArray(4)
while (currentCoroutineContext().isActive) {
// Read 4-byte length prefix
readFully(inp, lenBuf)
val len = (lenBuf[0].toInt() and 0xFF shl 24) or
(lenBuf[1].toInt() and 0xFF shl 16) or
(lenBuf[2].toInt() and 0xFF shl 8) or
(lenBuf[3].toInt() and 0xFF)
if (len <= 0 || len > 1_000_000) {
// The stream is desynchronised; skipping the prefix
// would parse payload bytes as the next length. Drop the
// connection and let the supervisor reconnect (I.14).
android.util.Log.w("EarhornCast", "bad length prefix $len — marking connection lost")
_connected.value = false
return
}
val msgBuf = ByteArray(len)
readFully(inp, msgBuf)
val decoded = CastMessageCodec.decode(msgBuf) ?: continue
handleMessage(decoded)
}
} catch (e: IOException) {
// Connection lost
_connected.value = false
} catch (t: Throwable) {
// Anything else — a payload the codec's guards let through, an
// unexpected type — used to kill this coroutine silently while
// the heartbeat kept the socket "connected": a mute socket of
// the screen's own making (review N.2, 2026-09-02). Treat it
// as a lost connection so the supervisor reconnects at once.
android.util.Log.w("EarhornCast", "read loop died (${t.javaClass.simpleName}: ${t.message}) — marking connection lost")
_connected.value = false
}
}
private fun readFully(inp: InputStream, buf: ByteArray) {
var offset = 0
while (offset < buf.size) {
val n = inp.read(buf, offset, buf.size - offset)
if (n < 0) throw IOException("End of stream")
offset += n
}
}
private suspend fun heartbeatLoop() {
while (currentCoroutineContext().isActive) {
delay(HEARTBEAT_INTERVAL_MS)
runCatching {
sendJson(RECEIVER_ID, NS_HEARTBEAT, """{"type":"PING"}""")
}
}
}
// ── Message handling ────────────────────────────────────────
private suspend fun handleMessage(msg: CastMessageCodec.DecodedMessage) {
val json = runCatching { JSONObject(msg.payloadUtf8) }.getOrNull()
?: return
val type = json.optString("type", "")
// Diagnostic — skip heartbeat to keep logcat readable, log everything else
// with a truncated payload preview. Filter on tag "EarhornCast" in
// logcat to see the receiver's message flow during a session. Useful
// when a device gets stuck on "Waiting for…" — the logs will show
// whether MEDIA_STATUS is arriving at all, and what it contains.
// Release logging: message TYPE only. Full-payload wire logging was
// the diag build's instrument (retired 2026-08-29 after the leader
// investigation got its specimens); type lines keep session
// narratives readable without the volume.
if (msg.namespace != NS_HEARTBEAT && type != "MEDIA_STATUS" && type != "RECEIVER_STATUS") {
android.util.Log.i("EarhornCast",
"rx ns=${msg.namespace.removePrefix("urn:x-cast:com.google.cast.")} type=$type")
}
// One compact line per media status: the receiver's own claim of
// state and position, straight off the wire — the forced-read
// record for stale-position and zombie investigations. This line
// was wrongly cut in the 0.4.3 trim; its absence cost the group
// its screen-side instrument for the Classical KC incident.
if (type == "MEDIA_STATUS") {
val st = _mediaStatus.value
android.util.Log.i("EarhornCast",
"status: state=${st.playerState} pos=${st.currentTime}s session=${st.mediaSessionId}")
}
when (msg.namespace) {
NS_HEARTBEAT -> {
if (type == "PING") {
sendJson(msg.sourceId, NS_HEARTBEAT, """{"type":"PONG"}""")
}
}
NS_RECEIVER -> {
if (type == "RECEIVER_STATUS") {
handleReceiverStatus(json)
}
}
NS_MEDIA -> {
when (type) {
"MEDIA_STATUS" -> handleMediaStatus(json)
"QUEUE_ITEM_IDS" -> handleQueueItemIds(json)
"QUEUE_ITEMS" -> handleQueueItems(json)
// Broadcast by the receiver whenever any sender edits
// the queue (insert/remove/reorder/replace). Refetch
// rather than applying the delta — queues here are
// small and a full refetch is unconditionally correct.
"QUEUE_CHANGE" -> scheduleQueueRefetch()
}
}
NS_CONNECTION -> {
if (type == "CLOSE") {
// Session ended
android.util.Log.i("EarhornCast", "transport session CLOSE from ${msg.sourceId}")
transportId = null
_mediaStatus.value = MediaStatus(statusSeq = _mediaStatus.value.statusSeq + 1)
clearQueueState()
}
}
}
}
private suspend fun handleReceiverStatus(json: JSONObject) {
val status = json.optJSONObject("status") ?: return
// Stamp the age here, before the no-app early return: an idle
// device otherwise never refreshed it and was polled every 10 s
// instead of every 25 (review N.3, 2026-09-02).
lastReceiverStatusAt = System.currentTimeMillis()
val apps = status.optJSONArray("applications")
if (apps == null || apps.length() == 0) {
// No media app running on the device. If we were attached to a
// session, this is POSITIVE evidence it is dead — and it is the
// only evidence we get on the freeze-then-recover path: the
// leader freezes, the socket dies, the reconnect supervisor
// succeeds once the device recovers, and the recovered device
// reports no application. Logging-and-returning here left the
// last PLAYING MediaStatus in place forever: sessionLive stuck
// true, hand-back structurally unable to fire, screens frozen
// on a stale queue (observed on three devices at once,
// 2026-08-28). Clear exactly as the CLOSE handler does.
if (transportId != null || _mediaStatus.value.mediaSessionId != 0) {
android.util.Log.i("EarhornCast",
"RECEIVER_STATUS reports no running app while we were " +
"attached — treating as session death (was tid=$transportId)")
transportId = null
_mediaStatus.value = MediaStatus(statusSeq = _mediaStatus.value.statusSeq + 1)
clearQueueState()
} else {
android.util.Log.i("EarhornCast",
"RECEIVER_STATUS: no media app running (not attached — no-op)")
}
return
}
val app = apps.getJSONObject(0)
val tid = app.optString("transportId", "")
val appId = app.optString("appId", "")
val displayName = app.optString("displayName", "")
if (tid.isBlank()) {
android.util.Log.i("EarhornCast", "RECEIVER_STATUS app has no transportId — appId=$appId displayName=$displayName")
return
}
// Extract volume (device volume, not a playback report — statusSeq
// is deliberately left unchanged by this copy).
val vol = status.optJSONObject("volume")
if (vol != null) {
_mediaStatus.value = _mediaStatus.value.copy(
volume = vol.optDouble("level", 1.0),
isMuted = vol.optBoolean("muted", false),
)
}
if (tid != transportId) {
android.util.Log.i("EarhornCast",
"attaching to transport tid=$tid appId=$appId displayName=$displayName")
transportId = tid
// Connect to the app's transport and get media status
sendJson(tid, NS_CONNECTION, """{"type":"CONNECT","origin":{}}""")
sendJson(tid, NS_MEDIA,
"""{"type":"GET_STATUS","requestId":${nextRequestId()}}""")
}
}
private fun handleMediaStatus(json: JSONObject) {
val statusArray = json.optJSONArray("status") ?: return
if (statusArray.length() == 0) {
// Empty status = idle. Don't wipe the previously-seen track metadata —
// a tap of "stop" briefly returns an empty status array before the next
// track loads, and the user shouldn't see the album art / title vanish
// for the millisecond between broadcasts.
_mediaStatus.value = _mediaStatus.value.let {
it.copy(playerState = "IDLE", currentTime = 0.0, statusSeq = it.statusSeq + 1)
}
return
}
val previous = _mediaStatus.value
val s = statusArray.getJSONObject(0)
// Always-present transport fields. Fall back to previous when the receiver
// omits a field rather than to a hardcoded default — receivers do send
// partial updates and we don't want a missing field to overwrite a known
// value with garbage.
val playerState = s.optString("playerState", previous.playerState)
.ifBlank { previous.playerState }
val currentTime = s.optDouble("currentTime", previous.currentTime)
val mediaSessionId = s.optInt("mediaSessionId", previous.mediaSessionId)
.let { if (it == 0) previous.mediaSessionId else it }
val supportedMediaCommands =
s.optLong("supportedMediaCommands", previous.supportedMediaCommands)
val repeatMode = s.optString("repeatMode").ifBlank { null } ?: previous.repeatMode
// Volume can ride on a status update even when there's no media block
// (the receiver layer exposes its own volume separately from the media
// app). Pull it before the media-block check so PAUSE/PLAY/SEEK updates
// can still propagate volume changes.
// ── Volume: DEVICE volume only ────────────────────────────────
// The media-namespace status carries the per-media STREAM volume,
// which sits at 1.0 essentially always — it is NOT the speaker's
// volume. The real device volume arrives via RECEIVER_STATUS on
// the receiver namespace (captured in the RECEIVER_STATUS handler
// and broadcast by the receiver whenever ANY sender or hardware
// control changes it). Previously this line overwrote the device
// volume with the stream volume on every media push, so observers
// read "100%" regardless of the actual speaker level — a first
// volume-up tap then sent an absolute 100% and blasted the room.
val newVolume = previous.volume
val newMuted = previous.isMuted
// ── Queue-relevant fields ─────────────────────────────────────
lastMediaStatusAt = System.currentTimeMillis()
val currentItemId = s.optInt("currentItemId", previous.currentItemId)
// New media session = new load = new queue. Clear the stale queue
// and fetch the fresh one. Also covers the initial connect (previous
// sessionId 0 → real id).
if (mediaSessionId != previous.mediaSessionId && mediaSessionId != 0) {
clearQueueState()
scheduleQueueRefetch()
}
// Receivers piggyback the current (and sometimes preloaded) queue
// items on MEDIA_STATUS. Merge them opportunistically so the
// now-playing row has metadata even before the batched fetch lands.
s.optJSONArray("items")?.let { inline ->
synchronized(queueLock) {
var merged = false
for (i in 0 until inline.length()) {
val qi = parseQueueItem(inline.optJSONObject(i) ?: continue) ?: continue
queueItemBuf[qi.itemId] = qi
merged = true
}
if (merged) publishQueueSnapshotLocked()
}
}
val media = s.optJSONObject("media")
if (media == null) {
// Transport-state-only update — PAUSE, RESUME, SEEK, or a heartbeat-ish
// refresh. The receiver does NOT redeliver the media block on every
// status push. Preserve the track metadata we already had; only update
// the volatile transport fields. THIS is the fix for the symptom where
// the displayed album art / title flickered back to a stale local-
// renderer track between Cast broadcasts.
_mediaStatus.value = previous.copy(
playerState = playerState,
currentTime = currentTime,
mediaSessionId = mediaSessionId,
supportedMediaCommands = supportedMediaCommands,
repeatMode = repeatMode,
volume = newVolume,
isMuted = newMuted,
currentItemId = currentItemId,
statusSeq = previous.statusSeq + 1,
)
return
}
// A media block IS present. Re-derive every metadata field from it, but
// treat empty strings as "not provided" and keep the previous value —
// `JSONObject.optString` returns "" (not null) for missing keys, and we
// don't want a partial metadata payload to blank the artist or album.
val duration = media.optDouble("duration", previous.duration)
val streamType = media.optString("streamType").ifBlank { null } ?: previous.streamType
val contentId = media.optString("contentId").ifBlank { null } ?: previous.contentId
val metadata = media.optJSONObject("metadata")
val title = metadata?.optString("title")?.ifBlank { null } ?: previous.title
val artist = (metadata?.optString("artist")?.ifBlank { null }
?: metadata?.optString("albumArtist")?.ifBlank { null })
?: previous.artist
val album = metadata?.optString("albumName")?.ifBlank { null } ?: previous.album
val images = metadata?.optJSONArray("images")
val albumArtUrl = if (images != null && images.length() > 0) {
images.getJSONObject(0).optString("url").ifBlank { null } ?: previous.albumArtUrl
} else {
previous.albumArtUrl
}
_mediaStatus.value = MediaStatus(
title = title,
artist = artist,
album = album,
albumArtUrl = albumArtUrl,
contentId = contentId,
duration = duration,
currentTime = currentTime,
playerState = playerState,
mediaSessionId = mediaSessionId,
supportedMediaCommands = supportedMediaCommands,
streamType = streamType,
repeatMode = repeatMode,
volume = newVolume,
isMuted = newMuted,
currentItemId = currentItemId,
statusSeq = previous.statusSeq + 1,
)
}
// ── Queue message handlers ───────────────────────────────────
private suspend fun handleQueueItemIds(json: JSONObject) {
val arr = json.optJSONArray("itemIds") ?: return
val ids = (0 until arr.length()).map { arr.optInt(it) }.filter { it != 0 }
android.util.Log.i("EarhornCast", "QUEUE_ITEM_IDS: ${ids.size} ids")
// Mutate under the lock; compute what to fetch; then suspend-send
// OUTSIDE the lock (never suspend while holding a monitor).
val toFetch = synchronized(queueLock) {
queueItemIdOrder = ids
// Drop buffered items that fell out of the queue, keep the rest
// so the list doesn't flash empty during a refetch. Surviving
// itemIds are trustworthy: item IDs are unique per insert, so an
// edited/replaced entry always arrives under a NEW id — there is
// no need to re-fetch metadata we already hold. (The previous
// "cheap safety" re-fetch of buffered items doubled QUEUE_GET_ITEMS
// traffic on every refetch; with several observers attached that
// measurably loads a group leader that is also doing multiroom.)
queueItemBuf.keys.retainAll(ids.toSet())
publishQueueSnapshotLocked()
ids.filterNot { queueItemBuf.containsKey(it) }
}
// Fetch missing metadata in batches of 20 — the SDK's MediaQueue
// uses the same chunk size to keep individual frames small.
toFetch.chunked(20).forEach { requestQueueItemsBatch(it) }
}
private fun handleQueueItems(json: JSONObject) {
val arr = json.optJSONArray("items") ?: return
var changed = false
synchronized(queueLock) {
for (i in 0 until arr.length()) {
val qi = parseQueueItem(arr.optJSONObject(i) ?: continue) ?: continue
queueItemBuf[qi.itemId] = qi
changed = true
}
if (changed) {
android.util.Log.i("EarhornCast",
"QUEUE_ITEMS batch → buf ${queueItemBuf.size}/${queueItemIdOrder.size}")
publishQueueSnapshotLocked()
}
}
}
/** Parse one queue item object: `{itemId, media: {contentId, duration,
* metadata: {title, artist|albumArtist, albumName, images: [{url}]}}}`. */
private fun parseQueueItem(obj: JSONObject): QueueItem? {
val itemId = obj.optInt("itemId", 0)
if (itemId == 0) return null
val media = obj.optJSONObject("media")
val metadata = media?.optJSONObject("metadata")
val images = metadata?.optJSONArray("images")
val artUrl = if (images != null && images.length() > 0) {
images.optJSONObject(0)?.optString("url")?.ifBlank { null }
} else null
return QueueItem(
itemId = itemId,
contentId = media?.optString("contentId")?.ifBlank { null },
title = metadata?.optString("title")?.ifBlank { null },
artist = metadata?.optString("artist")?.ifBlank { null }
?: metadata?.optString("albumArtist")?.ifBlank { null },
album = metadata?.optString("albumName")?.ifBlank { null },
albumArtUrl = artUrl,
durationSeconds = media?.optDouble("duration", 0.0) ?: 0.0,
)
}
/** Publish the ordered snapshot. Callers MUST hold [queueLock]. */
private fun publishQueueSnapshotLocked() {
if (queueItemIdOrder.isEmpty()) {
// No ID order yet (e.g. inline items merged before the first
// QUEUE_ITEM_IDS response) — publish buffered items by itemId
// order as a best-effort interim view.
_queueItems.value = queueItemBuf.values.sortedBy { it.itemId }
return
}
_queueItems.value = queueItemIdOrder.mapNotNull { queueItemBuf[it] }
}
/** Debounced full refetch — coalesces bursts of QUEUE_CHANGE broadcasts
* during bulk sender edits into a single round-trip. */
private fun scheduleQueueRefetch() {
val sc = scope ?: return
queueRefetchJob?.cancel()
queueRefetchJob = sc.launch {
delay(300)
runCatching { requestQueueItemIds() }
}
}
private fun clearQueueState() {
queueRefetchJob?.cancel()
queueRefetchJob = null
synchronized(queueLock) {
queueItemIdOrder = emptyList()
queueItemBuf.clear()
_queueItems.value = emptyList()
}
}
private fun nextRequestId(): Int = synchronized(this) { requestId++ }
// ── TLS trust (Chromecast uses self-signed certs) ───────────
private object PermissiveTrustManager : X509TrustManager {
override fun checkClientTrusted(chain: Array<java.security.cert.X509Certificate>?, authType: String?) {}
override fun checkServerTrusted(chain: Array<java.security.cert.X509Certificate>?, authType: String?) {}
override fun getAcceptedIssuers(): Array<java.security.cert.X509Certificate> = emptyArray()
}
}
/**
* Minimal hand-rolled protobuf codec for CastMessage.
*
* The CastMessage protobuf has exactly 7 fields, all simple types. Rather than
* pulling in the full protobuf-javalite dependency, we encode/decode the ~80 bytes
* of wire format directly. This keeps the client zero-dependency.
*
* Wire format reference:
* message CastMessage {
* required ProtocolVersion protocol_version = 1; // varint, always 0
* required string source_id = 2;
* required string destination_id = 3;
* required string namespace = 4;
* required PayloadType payload_type = 5; // varint, 0=STRING 1=BINARY
* optional string payload_utf8 = 6;
* optional bytes payload_binary = 7;
* }
*/
object CastMessageCodec {
data class DecodedMessage(
val sourceId: String,
val destinationId: String,
val namespace: String,
val payloadUtf8: String,
)
fun encode(
sourceId: String,
destinationId: String,
namespace: String,
payloadUtf8: String,
): ByteArray {
val buf = ByteArrayOutputStream(256)
// field 1: protocol_version = 0 (CASTV2_1_0), wire type 0 (varint)
writeVarint(buf, (1 shl 3) or 0)
writeVarint(buf, 0)
// field 2: source_id
writeString(buf, 2, sourceId)
// field 3: destination_id
writeString(buf, 3, destinationId)
// field 4: namespace
writeString(buf, 4, namespace)
// field 5: payload_type = 0 (STRING)
writeVarint(buf, (5 shl 3) or 0)
writeVarint(buf, 0)
// field 6: payload_utf8
writeString(buf, 6, payloadUtf8)
return buf.toByteArray()
}
fun decode(data: ByteArray): DecodedMessage? {
var sourceId = ""
var destinationId = ""
var namespace = ""
var payloadUtf8 = ""
var pos = 0
while (pos < data.size) {
val (tag, newPos) = readVarint(data, pos)
pos = newPos
val fieldNumber = tag ushr 3
val wireType = tag and 0x07
when (wireType) {
0 -> { // varint — skip the value
val (_, p) = readVarint(data, pos)
pos = p
}
2 -> { // length-delimited
val (len, p) = readVarint(data, pos)
pos = p
if (pos + len > data.size) return null
val bytes = data.copyOfRange(pos, pos + len)
pos += len
when (fieldNumber) {
2 -> sourceId = String(bytes, Charsets.UTF_8)
3 -> destinationId = String(bytes, Charsets.UTF_8)
4 -> namespace = String(bytes, Charsets.UTF_8)
6 -> payloadUtf8 = String(bytes, Charsets.UTF_8)
// 7 = payload_binary, ignored
}
}
else -> return null // unexpected wire type
}
}
return DecodedMessage(sourceId, destinationId, namespace, payloadUtf8)
}
private fun writeString(buf: ByteArrayOutputStream, fieldNumber: Int, value: String) {
val bytes = value.toByteArray(Charsets.UTF_8)
writeVarint(buf, (fieldNumber shl 3) or 2)
writeVarint(buf, bytes.size)
buf.write(bytes)
}
private fun writeVarint(buf: ByteArrayOutputStream, value: Int) {
var v = value
while (v > 0x7F) {
buf.write((v and 0x7F) or 0x80)
v = v ushr 7
}
buf.write(v and 0x7F)
}
private fun readVarint(data: ByteArray, startPos: Int): Pair<Int, Int> {
var result = 0
var shift = 0
var pos = startPos
while (pos < data.size) {
val b = data[pos].toInt() and 0xFF
result = result or ((b and 0x7F) shl shift)
pos++
if (b and 0x80 == 0) break
shift += 7
}
return result to pos
}
}
