package dev.stave.cast.raw
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withContext
import org.json.JSONObject
import java.io.BufferedInputStream
import java.io.BufferedOutputStream
import java.io.ByteArrayOutputStream
import java.io.IOException
import java.io.InputStream
import java.io.OutputStream
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 and works on libre builds, the
* GNU/Linux desktop, or anywhere with a JDK and a TLS-capable socket. Ported
* from the Android TV implementation: the protocol code is identical, only the
* logging surface differs (`android.util.Log` → stderr in the
* `[EarhornCast/Raw]` channel matching the existing `CastIntegration` style).
*
* Protocol references:
* - github.com/thibauts/node-castv2
* - github.com/home-assistant-libs/pychromecast
* - chromium.googlesource.com/.../cast_channel.proto
*/
class RawCastClient(
private 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
/** No bytes from the receiver for this long = the socket is dead.
* Receivers answer every PING with a PONG (and send their own PINGs),
* so a live connection never goes quiet this long. Without it a
* speaker that lost power or Wi-Fi (no FIN) kept `connected` true
* until the kernel gave up on TCP retransmits, about 15 minutes, and
* the observer's reconnect never ran. Same budget as pychromecast
* (10 s ping + 10 s pong wait). */
private const val RX_TIMEOUT_MS = 20_000L
private const val CONNECT_TIMEOUT_MS = 10_000
/** How long after a local [setVolume] incoming RECEIVER_STATUS volume
* readings are ignored. Long enough to outlive any status response
* that was already on the wire when the user clicked; short enough
* that genuine external changes (another sender's slider) surface
* within a poll cycle. */
private const val VOLUME_ECHO_SUPPRESS_MS = 1_500L
/** Diagnostic tag prefix. Matches `[EarhornCast]` from the sender-side
* integration so all cast-related output sits in one mental bucket. */
private const val TAG = "[EarhornCast/Raw]"
}
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,
val volume: Double = 1.0,
val isMuted: Boolean = false,
/** False until we have an authoritative DEVICE volume — from a
* RECEIVER_STATUS `volume` block, or latched by our own [setVolume].
* (MEDIA_STATUS never contributes: its `volume` is the media STREAM
* volume, a different knob — see [handleMediaStatus].) UI
* surfaces use this to distinguish "we don't know what the device
* is at" (do not display the slider as if it's the truth, do not
* apply optimistic updates) from "device reported 100%."
*
* The default volume of 1.0 above is a placeholder used until the
* first authoritative value arrives; without this flag the slider
* would sit at 100% on attach and a user click anywhere below
* that — say at 30% on a device actually at 5% — would jump the
* audible level by 6x in a single command, feeling like a
* "volume blast" even though the new value is correct. */
val volumeKnown: Boolean = false,
/** The receiver's authoritative "which queue item is playing" pointer —
* the `currentItemId` top-level field of MEDIA_STATUS. Cross-referenced
* against the QUEUE_ITEM_IDS order to derive the queue index. NEVER
* count the index locally; this value cannot drift from what's audible.
* Null until the receiver first reports it (or when the session has no
* receiver-side queue at all — e.g. a single-item load). Sticky across
* transport-only pushes that omit it, same as the metadata fields. */
val currentItemId: Int? = null,
/** Advances by one for every MEDIA_STATUS the receiver sends, and for
* nothing else. RECEIVER_STATUS (device volume) copies keep it. The
* observer keys "is this a new playback report?" on this number
* (ported from Android, 2026-09-04: a volume report used to count as
* "same position twice" and hold the bar). */
val statusSeq: Long = 0L,
/** Receiver queue repeat mode: REPEAT_OFF, REPEAT_ALL, REPEAT_SINGLE,
* REPEAT_ALL_AND_SHUFFLE; "" until reported. */
val repeatMode: String = "",
)
// ── Receiver-queue introspection (follower queue) ────────────
//
// The receiver is the source of truth for its queue and answers
// introspection requests from any connected peer — the same mechanism
// Google's own MediaQueue sender class uses. We expose the wire traffic
// as a stream of [QueueEvent]s; RawCastObserver owns the assembled
// queue state (ordering, item buffer, batching, debouncing).
/** One queue entry as it appears in a QUEUE_ITEMS response (or the `items`
* array some receivers piggyback on MEDIA_STATUS pushes). Blank strings
* from lax JSON parsers are normalised to null before this is built. */
data class QueueItemLite(
val itemId: Int,
val contentId: String?,
val title: String?,
val artist: String?,
val album: String?,
val albumArtUrl: String?,
val durationSeconds: Double?,
)
sealed interface QueueEvent {
/** QUEUE_ITEM_IDS response: the receiver's full ordered id list. */
data class ItemIds(val itemIds: List<Int>) : QueueEvent
/** QUEUE_ITEMS response batch — or items piggybacked on a MEDIA_STATUS
* push ([fromStatusPush] = true). Piggybacked payloads carry only the
* current (and sometimes preloaded) item; merge them opportunistically
* so the now-playing row has metadata before the batched fetch lands,
* but never treat them as a complete queue. */
data class Items(
val items: List<QueueItemLite>,
val fromStatusPush: Boolean = false,
) : QueueEvent
/** QUEUE_CHANGE broadcast — the receiver announcing that *some* sender
* edited the queue. [changeType] is ITEMS_CHANGED|INSERT|REMOVE|UPDATE.
* The correct reaction is a (debounced) refetch of the id list. */
data class Change(val changeType: String, val itemIds: List<Int>) : QueueEvent
/** mediaSessionId transitioned to a new non-zero value — a new load,
* which means a new receiver-side queue. Everything buffered for the
* old session is invalid. */
data class SessionChanged(val mediaSessionId: Int) : QueueEvent
}
private val _queueEvents = MutableSharedFlow<QueueEvent>(
// Queue traffic is bursty (batched QUEUE_ITEMS during a refetch) and
// every event is recoverable by a fresh refetch, so a bounded buffer
// with drop-oldest is the right failure mode — never suspend the read
// loop on a slow collector.
extraBufferCapacity = 256,
onBufferOverflow = kotlinx.coroutines.channels.BufferOverflow.DROP_OLDEST,
)
/** Wire-level queue traffic. Collected by [RawCastObserver]'s queue manager. */
val queueEvents: SharedFlow<QueueEvent> = _queueEvents.asSharedFlow()
private val _mediaStatus = MutableStateFlow(MediaStatus())
val mediaStatus: StateFlow<MediaStatus> = _mediaStatus.asStateFlow()
private val _connected = MutableStateFlow(false)
val connected: StateFlow<Boolean> = _connected.asStateFlow()
// Written by connect()/disconnect() and the reader thread, read by command
// callers on other threads, hence @Volatile.
@Volatile private var socket: SSLSocket? = null
@Volatile private var output: OutputStream? = null
@Volatile private var input: InputStream? = null
@Volatile private var scope: CoroutineScope? = null
@Volatile private var transportId: String? = null
/** Wall-clock of the last frame read from the socket (any namespace). */
@Volatile private var lastRxAt: Long = 0L
private var requestId = 1
/** Wall-clock of the last RECEIVER_STATUS, idle or not (ported from Android). */
@Volatile var lastReceiverStatusAt: Long = 0L
private set
/** Log every non-heartbeat message with a payload preview. Off by default:
* one line per status, every few seconds, forever, on every desktop. */
@Volatile var verboseWireLog: Boolean = false
/** Wallclock of the most recent local [setVolume]. RECEIVER_STATUS volume
* ingest is suppressed for [VOLUME_ECHO_SUPPRESS_MS] after this so a
* status that was already in flight when the user clicked (carrying the
* OLD device level) can't briefly bounce the slider back before the
* post-set broadcast confirms the new value. Written on the caller's
* coroutine, read on the socket-reader coroutine — hence @Volatile. */
@Volatile
private var localVolumeSetAtMs = 0L
// ── Connect / Disconnect ─────────────────────────────────────
suspend fun connect() = withContext(Dispatchers.IO) {
disconnect()
val sc = CoroutineScope(SupervisorJob() + Dispatchers.IO)
scope = sc
System.err.println("$TAG 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
// Published before connecting so a disconnect() from another thread
// (the user stops following mid-connect) closes it and aborts the
// blocking connect/handshake instead of waiting for the timeouts.
socket = sock
try {
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. Reset after — the read loop
// blocks on purpose.
sock.soTimeout = CONNECT_TIMEOUT_MS
sock.startHandshake()
sock.soTimeout = 0
output = BufferedOutputStream(sock.outputStream)
input = BufferedInputStream(sock.inputStream)
} catch (t: Throwable) {
// A timeout or TLS failure used to leave the socket's fd open,
// once per reconnect attempt.
runCatching { sock.close() }
if (socket === sock) socket = null
sc.cancel()
if (scope === sc) scope = null
throw t
}
lastRxAt = System.currentTimeMillis()
_connected.value = true
System.err.println("$TAG TLS handshake OK — sending CONNECT + GET_STATUS")
// Start reader + heartbeat
sc.launch { readLoop(sock) }
sc.launch { heartbeatLoop(sock) }
// 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) {
System.err.println("$TAG disconnecting from $host:$port")
}
scope?.cancel()
scope = null
runCatching { socket?.close() }
socket = null
output = null
input = null
transportId = null
_connected.value = false
}
// ── 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) {
// Optimistically reflect the new level in our own MediaStatus BEFORE
// sending. The observer bridges _mediaStatus → PlaybackState on every
// emission; MEDIA_STATUS payloads never contribute volume (their
// `volume` field is the media STREAM volume — see the note in
// [handleMediaStatus]), so this optimistic write is the baseline every
// subsequent media update preserves. An authoritative RECEIVER_STATUS
// still wins if the device reports something different — but only
// after [VOLUME_ECHO_SUPPRESS_MS], so a status already in flight with
// the OLD level can't bounce the slider before the post-set broadcast
// lands (see [handleReceiverStatus]).
val clamped = level.coerceIn(0.0, 1.0)
localVolumeSetAtMs = System.currentTimeMillis()
_mediaStatus.value = _mediaStatus.value.copy(volume = clamped, volumeKnown = true)
sendJson(
RECEIVER_ID, NS_RECEIVER,
"""{"type":"SET_VOLUME","volume":{"level":$clamped},"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()}}""",
)
}
/**
* Re-request the device's RECEIVER_STATUS. Unlike [requestMediaStatus], this
* goes to the receiver-platform namespace and reliably returns a payload
* with the `volume` sub-object — MEDIA_STATUS broadcasts often omit it
* after the initial connect because volume changes are infrequent. The
* observer uses this on attach (with backoff) to bootstrap an
* authoritative volume reading; without it the slider sits on the
* default 100% until something happens to make the receiver re-broadcast,
* which on a steady-state PLAYING session may never happen.
*/
suspend fun requestReceiverStatus() {
sendJson(
RECEIVER_ID, NS_RECEIVER,
"""{"type":"GET_STATUS","requestId":${nextRequestId()}}""",
)
}
// ── Queue introspection commands ─────────────────────────────
//
// All three go to the app transportId (NOT receiver-0) on the media
// namespace, and all require a live mediaSessionId from MEDIA_STATUS —
// the guards below silently no-op until both are known, which matches
// how transport commands already behave.
/** Ask the receiver for its ordered queue-item id list. The response
* arrives asynchronously as a [QueueEvent.ItemIds]. */
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()}}""",
)
}
/** Fetch metadata for a batch of queue item ids. Callers chunk to the
* shared batch size (20); the response arrives as [QueueEvent.Items]. */
suspend fun requestQueueItems(itemIds: List<Int>) {
if (itemIds.isEmpty()) return
val tid = transportId ?: return
val msid = _mediaStatus.value.mediaSessionId
if (msid == 0) return
val ids = itemIds.joinToString(",")
sendJson(
tid, NS_MEDIA,
"""{"type":"QUEUE_GET_ITEMS","mediaSessionId":$msid,"itemIds":[$ids],"requestId":${nextRequestId()}}""",
)
}
/**
* Jump the receiver's playback position to [itemId] — position-only,
* queue contents untouched. This is the ONLY queue mutation a follower
* is allowed to send: the driving sender's local queue is authoritative
* and (on Android at least) it full-reloads the receiver queue on its
* own edits, so a follower REMOVE/REORDER would be silently clobbered
* and desync the sender in the interim. QUEUE_UPDATE with currentItemId
* is safe because the sender's index tracker follows the receiver via
* its own status callbacks.
*/
/** Set the receiver queue's repeat mode (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 (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 queueJumpTo(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) {
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 sock = socket
val out = output ?: return@withContext
// Ported from Android: a write on a dead socket (receiver rebooted,
// network drop) is the earliest sign the connection is gone, since
// the read loop may block forever on a half-open socket. Mark the
// connection lost so the observer's reconnect supervisor runs, and
// don't throw into the caller (commands are fire-and-forget).
try {
synchronized(out) {
out.write(lenBuf)
out.write(msg)
out.flush()
}
} catch (e: IOException) {
System.err.println("$TAG send failed (${e.message}) — marking connection lost")
markLost(sock)
}
}
/** Flag the connection lost, but only if [sock] is still the current
* socket: a reader or writer of a socket that a reconnect has already
* replaced must not knock the new connection's flag back to false. */
private fun markLost(sock: SSLSocket?) {
if (sock != null && socket === sock) _connected.value = false
}
private suspend fun readLoop(sock: SSLSocket) {
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) {
// Desynchronised stream: skipping the prefix would parse
// payload bytes as the next length. Drop the connection.
System.err.println("$TAG bad length prefix $len — marking connection lost")
markLost(sock)
return
}
val msgBuf = ByteArray(len)
readFully(inp, msgBuf)
lastRxAt = System.currentTimeMillis()
val decoded = CastMessageCodec.decode(msgBuf) ?: continue
handleMessage(decoded)
}
} catch (e: IOException) {
// Connection lost (or closed by disconnect(); markLost ignores
// a socket that has already been replaced).
markLost(sock)
} catch (e: kotlinx.coroutines.CancellationException) {
throw e
} catch (t: Throwable) {
// Anything else used to kill this coroutine silently while the
// heartbeat kept the socket "connected" — a mute socket.
System.err.println("$TAG read loop died (${t.javaClass.simpleName}: ${t.message}) — marking connection lost")
markLost(sock)
}
}
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(sock: SSLSocket) {
while (currentCoroutineContext().isActive) {
delay(HEARTBEAT_INTERVAL_MS)
if (socket !== sock) return
// Rx watchdog: a socket that has gone silent past RX_TIMEOUT_MS is
// dead even though nothing has told us so (no FIN, read blocked).
// Closing it unblocks the read loop; marking it lost starts the
// observer's reconnect supervisor now rather than in ~15 minutes.
val quietMs = System.currentTimeMillis() - lastRxAt
if (quietMs > RX_TIMEOUT_MS) {
System.err.println("$TAG nothing received for ${quietMs / 1000}s — marking connection lost")
markLost(sock)
runCatching { sock.close() }
return
}
// sendJson handles IOException itself (marks the connection lost).
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 output readable; log everything
// else with a truncated payload preview. Filter on "[EarhornCast/Raw]"
// in stderr to see the receiver's message flow during a session. Useful
// when a device gets stuck on "Waiting for…" — the output will show
// whether MEDIA_STATUS is arriving at all, and what it contains.
if (verboseWireLog && msg.namespace != NS_HEARTBEAT) {
val preview = msg.payloadUtf8.take(280) +
if (msg.payloadUtf8.length > 280) "…" else ""
System.err.println(
"$TAG rx ns=${msg.namespace.removePrefix("urn:x-cast:com.google.cast.")} " +
"type=$type payload=$preview",
)
}
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" -> {
val arr = json.optJSONArray("itemIds")
val ids = if (arr == null) emptyList() else {
(0 until arr.length()).mapNotNull { i ->
arr.optInt(i, -1).takeIf { it > 0 }
}
}
_queueEvents.tryEmit(QueueEvent.ItemIds(ids))
}
"QUEUE_ITEMS" -> {
val arr = json.optJSONArray("items")
if (arr != null) {
val items = (0 until arr.length()).mapNotNull { i ->
parseQueueItem(arr.optJSONObject(i))
}
if (items.isNotEmpty()) {
_queueEvents.tryEmit(QueueEvent.Items(items))
}
}
}
"QUEUE_CHANGE" -> {
val changeType = json.optString("changeType", "ITEMS_CHANGED")
val arr = json.optJSONArray("itemIds")
val ids = if (arr == null) emptyList() else {
(0 until arr.length()).mapNotNull { i ->
arr.optInt(i, -1).takeIf { it > 0 }
}
}
_queueEvents.tryEmit(QueueEvent.Change(changeType, ids))
}
}
}
NS_CONNECTION -> {
if (type == "CLOSE") {
// Session ended
System.err.println("$TAG transport session CLOSE from ${msg.sourceId}")
transportId = null
// Keep the DEVICE volume reading: it belongs to the
// receiver, not the session that just ended. Resetting it
// locked the follower's volume slider (volumeKnown=false)
// until the next RECEIVER_STATUS with a volume block.
val prev = _mediaStatus.value
_mediaStatus.value = MediaStatus(
volume = prev.volume,
isMuted = prev.isMuted,
volumeKnown = prev.volumeKnown,
statusSeq = prev.statusSeq + 1,
)
}
}
}
}
private suspend fun handleReceiverStatus(json: JSONObject) {
val status = json.optJSONObject("status") ?: return
lastReceiverStatusAt = System.currentTimeMillis()
val apps = status.optJSONArray("applications")
if (apps == null || apps.length() == 0) {
// No media app running. Now that the observer polls NS_RECEIVER
// every 3s (for device volume), this branch fires continuously on
// an idle device — so it logs and tears down only on the
// TRANSITION (we previously had a transport, now the app is
// gone), and stays silent while already idle. Clearing
// [transportId] mirrors the CLOSE handler: the old transport is
// dead, and nulling it turns subsequent media polls/commands
// into no-ops instead of writes to a defunct session. Track
// metadata stays sticky (art/title shouldn't vanish); only the
// transport state drops to IDLE, same as the empty-status-array
// branch of [handleMediaStatus].
// Also when transportId is already null but the last status still
// claims a live session: after a reconnect, connect() has cleared
// transportId, and a recovered device that reports no app is the
// only evidence the old session is gone (ported from Android's
// 2026-08-28 fix). Without this a stale PLAYING state stuck until
// the observer's 150 s silence watchdog.
val ms0 = _mediaStatus.value
if (transportId != null || (ms0.mediaSessionId != 0 && ms0.playerState != "IDLE")) {
System.err.println(
"$TAG RECEIVER_STATUS reports no running media app — session ended; detaching transport",
)
transportId = null
_mediaStatus.value = _mediaStatus.value.let {
it.copy(playerState = "IDLE", currentTime = 0.0, statusSeq = it.statusSeq + 1)
}
}
return
}
// Pick the app that actually owns the media namespace, not just apps[0].
// CCA groups in particular run TWO apps simultaneously when a session is
// active: the "Multizone Leader" (managing audio sync across followers,
// exposes its own namespaces but NOT media), and the sender's actual
// receiver app (Default Media Receiver, Spotify, etc — exposes
// urn:x-cast:com.google.cast.media). Attaching to the leader gives us a
// working channel that never delivers MEDIA_STATUS payloads, which is
// why the UI was sitting at "Nothing playing" while the user could hear
// music: transport state was syncing via the receiver's other channels,
// but media metadata only rides on the media namespace.
//
// Fallback to apps[0] if no app exposes the media namespace (e.g., the
// device is showing only a leader or a non-media app) — preserves the
// pre-fix behavior in the unlikely edge case rather than silently doing
// nothing.
val app = (0 until apps.length()).map { apps.getJSONObject(it) }
.firstOrNull { candidate ->
val ns = candidate.optJSONArray("namespaces") ?: return@firstOrNull false
(0 until ns.length()).any { i ->
ns.getJSONObject(i).optString("name") == NS_MEDIA
}
} ?: apps.getJSONObject(0)
val tid = app.optString("transportId", "")
val appId = app.optString("appId", "")
val displayName = app.optString("displayName", "")
if (tid.isBlank()) {
System.err.println(
"$TAG RECEIVER_STATUS app has no transportId — appId=$appId displayName=$displayName",
)
return
}
// Extract DEVICE volume. RECEIVER_STATUS is the one place the real
// device/group level lives (MEDIA_STATUS only ever carries the media
// STREAM volume — see [handleMediaStatus]). Skip the copy inside the
// short post-[setVolume] window: a GET_STATUS response that was
// already in flight when the user clicked still reports the OLD
// level, and ingesting it would bounce the slider back for a beat
// before the volume-change broadcast confirms the new value. After
// the window the receiver is authoritative again, which is what lets
// volume changes made by OTHER senders (the phone's slider) track.
val vol = status.optJSONObject("volume")
val sinceLocalSet = System.currentTimeMillis() - localVolumeSetAtMs
if (vol != null && sinceLocalSet > VOLUME_ECHO_SUPPRESS_MS) {
_mediaStatus.value = _mediaStatus.value.copy(
volume = vol.optDouble("level", 1.0),
isMuted = vol.optBoolean("muted", false),
volumeKnown = true,
)
}
if (tid != transportId) {
System.err.println(
"$TAG attaching to media-app 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 }
// New load = new receiver-side queue. Tell the queue manager so it can
// clear everything buffered for the old session and refetch. The very
// first MEDIA_STATUS after connect also lands here (previous id is 0),
// which doubles as the initial fetch trigger.
if (mediaSessionId != 0 && mediaSessionId != previous.mediaSessionId) {
_queueEvents.tryEmit(QueueEvent.SessionChanged(mediaSessionId))
}
// Receiver-authoritative queue pointer. Sticky across pushes that omit
// it (transport-only refreshes drop it along with the media block).
val currentItemId = s.optInt("currentItemId", -1).takeIf { it > 0 }
?: previous.currentItemId
// Receivers piggyback an `items` array on some MEDIA_STATUS pushes
// containing the current (and sometimes preloaded) item. Merge those
// opportunistically so the now-playing row has metadata before the
// batched QUEUE_GET_ITEMS fetch lands.
s.optJSONArray("items")?.let { arr ->
val items = (0 until arr.length()).mapNotNull { i ->
parseQueueItem(arr.optJSONObject(i))
}
if (items.isNotEmpty()) {
_queueEvents.tryEmit(QueueEvent.Items(items, fromStatusPush = true))
}
}
// ── Volume: deliberately IGNORED here ────────────────────────────
// MEDIA_STATUS *does* sometimes carry a `volume` sub-object, but per
// the Cast protocol that is the media STREAM volume (the per-item
// attenuation a sender can apply on top of the device level), NOT the
// device/group volume the user hears and controls. The Default Media
// Receiver and CCA groups report the stream volume pegged at 1.0
// essentially always. An earlier revision ingested it as if it were
// the device volume, which produced the "set the slider to 40, watch
// it snap back to 100 a few seconds later while the audio stays at
// 40" bug: the user's SET_VOLUME (receiver namespace → device volume)
// worked, then the next 3s media poll delivered volume:{level:1.0}
// (stream volume) and clobbered the display.
//
// Device volume flows exclusively through RECEIVER_STATUS (see
// [handleReceiverStatus]) plus the optimistic latch in [setVolume].
// Both status constructions below carry `previous.*` forward
// untouched.
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,
currentItemId = currentItemId,
repeatMode = s.optString("repeatMode").ifBlank { null } ?: previous.repeatMode,
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 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")
// Keep the previous picture only while the SAME item is loaded (a
// partial status). A different item with no picture gets none — a
// radio station used to wear the last album's cover (decision,
// 2026-09-07).
val sameItem = contentId == previous.contentId
val albumArtUrl = if (images != null && images.length() > 0) {
images.getJSONObject(0).optString("url").ifBlank { null } ?: previous.albumArtUrl.takeIf { sameItem }
} else {
previous.albumArtUrl.takeIf { sameItem }
}
_mediaStatus.value = MediaStatus(
title = title,
artist = artist,
album = album,
albumArtUrl = albumArtUrl,
contentId = contentId,
duration = duration,
currentTime = currentTime,
playerState = playerState,
mediaSessionId = mediaSessionId,
volume = previous.volume,
isMuted = previous.isMuted,
volumeKnown = previous.volumeKnown,
currentItemId = currentItemId,
repeatMode = s.optString("repeatMode").ifBlank { null } ?: previous.repeatMode,
statusSeq = previous.statusSeq + 1,
)
}
/**
* Parse one queue item from a QUEUE_ITEMS response or a MEDIA_STATUS
* `items` piggyback. Per the cross-platform spec: `artist` sometimes
* arrives as `albumArtist` instead — check both; missing keys come back
* as empty strings from lax JSON parsers — treat blank as absent.
*/
private fun parseQueueItem(o: JSONObject?): QueueItemLite? {
if (o == null) return null
val itemId = o.optInt("itemId", -1)
if (itemId <= 0) return null
val media = o.optJSONObject("media")
val metadata = media?.optJSONObject("metadata")
val images = metadata?.optJSONArray("images")
val duration = media?.optDouble("duration", Double.NaN)
?.takeIf { it.isFinite() && it > 0.0 }
return QueueItemLite(
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 = if (images != null && images.length() > 0) {
images.optJSONObject(0)?.optString("url")?.ifBlank { null }
} else null,
durationSeconds = duration,
)
}
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
// A varint that overflows to a negative length passed the
// old `pos + len > size` check and threw in copyOfRange,
// which dropped the connection.
if (len < 0 || len > data.size - pos) 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
}
}
