package dev.stave.upnp.renderer
import dev.stave.domain.model.DidlItem
import dev.stave.domain.model.RemoteRenderer
import dev.stave.domain.renderer.PlaybackState
import dev.stave.domain.renderer.RepeatMode
import dev.stave.domain.renderer.Renderer
import dev.stave.upnp.UpnpServiceHolder
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
import kotlinx.coroutines.suspendCancellableCoroutine
import kotlinx.coroutines.withContext
import org.jupnp.controlpoint.ActionCallback
import org.jupnp.controlpoint.SubscriptionCallback
import org.jupnp.model.action.ActionException
import org.jupnp.model.action.ActionInvocation
import org.jupnp.model.gena.CancelReason
import org.jupnp.model.gena.GENASubscription
import org.jupnp.model.message.UpnpResponse
import org.jupnp.model.meta.RemoteDevice
import org.jupnp.model.meta.RemoteService
import org.jupnp.model.types.UDAServiceType
import org.jupnp.support.avtransport.lastchange.AVTransportLastChangeParser
import org.jupnp.support.avtransport.lastchange.AVTransportVariable
import org.jupnp.support.lastchange.Event
import org.jupnp.support.lastchange.EventedValue
import org.jupnp.support.renderingcontrol.lastchange.RenderingControlLastChangeParser
import org.jupnp.support.renderingcontrol.lastchange.RenderingControlVariable
import kotlin.coroutines.resume
import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
/**
* A [Renderer] implementation that drives a REMOTE UPnP MediaRenderer device discovered on
* the LAN. The remote device is the actual audio source (another Earhorn instance, a Linn
* streamer, gmediarender, BubbleUPnP receiver, etc.) — this class is the control-point
* adapter that makes a remote renderer look like a local one to the rest of the app.
*
* Two halves of behavior:
*
* 1. **Outgoing actions.** [play], [pause], [stop], [seekTo], [next], [previous],
* [setVolume] each translate to a UPnP action invoked on the remote device's
* AVTransport or RenderingControl service. We use jUPnP's `ControlPoint.execute`
* with [ActionCallback], wrapped in [suspendCancellableCoroutine] so callers can
* `suspend` on completion the same way they do with the local renderer.
*
* 2. **Incoming state.** We open GENA subscriptions to the remote device's AVTransport
* and RenderingControl `LastChange` evented state variables. Push notifications
* from the renderer arrive as [eventReceived] callbacks; we parse the LastChange
* payload with the service-specific parser and update [state] accordingly. Plus
* a fallback poller for position (which UPnP doesn't event — it's pull-only via
* `GetPositionInfo`).
*
* **The queue is ours.** UPnP AVTransport has no protocol-level queue, so this
* class keeps the queue and feeds the device one track at a time
* (SetAVTransportURI), pre-loading the next one with SetNextAVTransportURI for
* gapless playback. Two ways to move on when a track ends, as on Android:
*
* - **Gapless:** the device starts the pre-loaded next track itself and reports
* the new `CurrentTrackURI` ([handleCurrentTrackUriChange]).
* - **Fallback:** many devices ignore SetNext (TVs, some Cast bridges,
* gmediarender builds), and GENA events may never arrive (a firewall on this
* computer is enough). The poller asks for the transport state every second
* while playing and every ten seconds otherwise (Android parity
* 2026-09-02), and when the device goes STOPPED within a few seconds of the
* end of OUR track we send the next one ([onRemoteStopped]). Any other stop
* is a stop (Android parity, 2026-09-02).
*
* Next / Previous, repeat and shuffle all act on our queue (Android parity, 2026-10-29,
* Android): the device only ever holds one track, so its own Next or REPEAT_ALL
* would do nothing useful.
*
* **Lifecycle.**
*
* The coordinator (next iteration) constructs a fresh [RemoteUpnpRenderer] each time
* the user picks a different remote renderer, calls [attach] to open subscriptions,
* and calls [detach] when switching away or shutting down. [detach] cancels the
* subscription and stops the position poller cleanly so we don't leak GENA listeners
* on the remote device — even Earhorn-to-Earhorn subscriptions don't auto-clean up
* if the controller process dies abruptly, so detaching is hygienic for graceful
* shutdown.
*/
class RemoteUpnpRenderer(
private val target: RemoteRenderer,
private val holder: UpnpServiceHolder,
) : Renderer {
/** UDN of the remote renderer this instance is bound to. Public so the
* coordinator can match an attached instance against the user's selection
* without holding a separate lookup table. Stable for the life of the
* instance — to drive a different renderer, build a new instance. */
val targetUdn: String get() = target.udn
/** Display name of the remote — likewise public for the same reason. */
val targetFriendlyName: String get() = target.friendlyName
private val _state = MutableStateFlow(PlaybackState())
override val state: StateFlow<PlaybackState> = _state.asStateFlow()
// ---- Queue state --------------------------------------------------------
/**
* Local queue maintained on the control-point side. UPnP AVTransport has no
* queue concept at the protocol level — `SetAVTransportURI` loads one track
* and `SetNextAVTransportURI` pre-loads one more, that's the entire surface.
* To support multi-track albums / playlists / "queue this after current",
* the control point (us) owns queue state and feeds tracks to the renderer
* one at a time, using `SetNextAVTransportURI` for gapless transitions.
*
* Every read and write happens on [queueCtx]: user actions, GENA events and
* the poller all used to touch it from different threads, and a track
* ending while the user removed a row could throw out of `queue.toList()`.
*/
private val queue: MutableList<DidlItem.Track> = mutableListOf()
private var queueIndex: Int = -1
/** One thread at a time for the queue and the advance bookkeeping below.
* Network calls suspend and free it, so calls can interleave between
* steps, but no two non-suspending sections ever run together. FIFO, so
* GENA events are handled in the order they arrived. */
private val queueCtx = Dispatchers.IO.limitedParallelism(1)
/** True after a Stop from anywhere (ours, the device's buttons, another
* app): the next STOPPED must not advance. Cleared by [play]. */
@Volatile private var userRequestedStop = false
/** When we last sent a track (or saw the device move to our next one). A
* STOPPED within [SEND_GRACE_MS] of that is the device switching tracks,
* never "our track ended" (Android parity, 2026-09-02). */
@Volatile private var lastTrackSentAt = 0L
/** Position seen at the last poll while playing; decides whether a STOPPED
* means "our track finished" or "someone stopped it". */
@Volatile private var lastSeenPosition: Duration = Duration.ZERO
/** URI of the track we sent most recently, and what the device calls it if
* it reports a rewritten form (a proxy or a redirect) — see
* [handleCurrentTrackUriChange]. The previous pair lets a late echo of the
* track we just replaced be recognised as stale instead of a hijack. */
private var currentRemoteUri: String? = null
private var currentRemoteAlias: String? = null
private var previousRemoteUri: String? = null
private var previousRemoteAlias: String? = null
/** Queue order before shuffle was turned on; restored when it goes off. */
private var preShuffleOrder: List<DidlItem.Track>? = null
/**
* URI we just told the renderer to play (via SetAVTransportURI). Used to
* suppress the auto-advance detection logic in [parseAvtLastChange] —
* when the renderer echoes back the URI we just set, we don't want to
* interpret that as an auto-advance event.
*/
@Volatile
private var pendingExpectedUri: String? = null
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
/** Subscription tickets we need to cancel on detach. jUPnP's
* [SubscriptionCallback.end] sends the UNSUBSCRIBE request to the renderer. */
private var avtSubscription: SubscriptionCallback? = null
private var rcSubscription: SubscriptionCallback? = null
/** Position poller. UPnP doesn't event the current track position — it's only
* queryable via [GetPositionInfo]. We poll once a second while we've seen a
* PLAYING transport state, and stop while paused/stopped to avoid network noise. */
private var positionPollerJob: Job? = null
// ---- Lifecycle ----------------------------------------------------------
/**
* Open the GENA subscriptions and prime [state] with an initial snapshot. Idempotent;
* calling twice is a no-op. Doesn't throw — failures are logged and the renderer
* comes up in a degraded "connected but no events" mode that callers can still issue
* actions against (each action runs an independent invocation).
*/
fun attach() {
if (avtSubscription != null) return // already attached
log("attach() target=${target.friendlyName} udn=${target.udn}")
val device = findRemoteDevice()
if (device == null) {
log("Device with UDN ${target.udn} is not in the registry — cannot attach")
return
}
// Initial best-effort state pull via individual GetXxx actions. The renderer
// will then keep us in sync via GENA events. If GENA fails (some bargain-bin
// renderers don't fire events even though they accept the subscription), the
// initial snapshot is at least useful and the action surface still works.
scope.launch(queueCtx) {
catching("pullInitialAvtState") { pullInitialAvtState(device) }
catching("pullInitialVolume") { pullInitialVolume(device) }
}
device.findAvtService()?.let { svc ->
val cb = object : SubscriptionCallback(svc) {
override fun established(s: GENASubscription<*>) {
log("AVT subscription established")
}
override fun ended(
s: GENASubscription<*>,
reason: CancelReason?,
resp: UpnpResponse?,
) {
log("AVT subscription ended reason=$reason rc=${resp?.statusCode}")
}
override fun eventReceived(s: GENASubscription<*>) {
val raw = s.currentValues["LastChange"]?.value as? String ?: return
scope.launch(queueCtx) { onAvtLastChange(raw) }
}
override fun eventsMissed(s: GENASubscription<*>, count: Int) {
log("AVT subscription missed $count event(s) — state may be stale")
}
override fun failed(
s: GENASubscription<*>?,
resp: UpnpResponse?,
ex: Exception?,
defaultMsg: String,
) {
log("AVT subscription failed: $defaultMsg ${ex?.message ?: ""}")
}
}
holder.controlPoint?.execute(cb)
avtSubscription = cb
} ?: log("Remote device has no AVTransport service — degraded mode")
device.findRcService()?.let { svc ->
val cb = object : SubscriptionCallback(svc) {
override fun established(s: GENASubscription<*>) {}
override fun ended(
s: GENASubscription<*>,
reason: CancelReason?,
resp: UpnpResponse?,
) {}
override fun eventReceived(s: GENASubscription<*>) {
val raw = s.currentValues["LastChange"]?.value as? String ?: return
scope.launch(queueCtx) { parseRcLastChange(raw) }
}
override fun eventsMissed(s: GENASubscription<*>, count: Int) {}
override fun failed(
s: GENASubscription<*>?,
resp: UpnpResponse?,
ex: Exception?,
defaultMsg: String,
) {
log("RC subscription failed: $defaultMsg")
}
}
holder.controlPoint?.execute(cb)
rcSubscription = cb
} ?: log("Remote device has no RenderingControl service — volume disabled")
startPositionPoller()
}
/** Cancel GENA subscriptions, stop pollers, and release the coroutine scope.
* Safe to call multiple times; subsequent calls are no-ops. */
fun detach() {
log("detach() target=${target.friendlyName}")
runCatching { avtSubscription?.end() }
runCatching { rcSubscription?.end() }
avtSubscription = null
rcSubscription = null
positionPollerJob?.cancel()
positionPollerJob = null
scope.cancel()
}
// ---- Outgoing actions (Renderer interface) ------------------------------
override suspend fun play() {
userRequestedStop = false
invokeAvtAction("Play", mapOf("Speed" to "1"))
}
override suspend fun pause() {
invokeAvtAction("Pause")
}
override suspend fun stop(): Unit = withContext(queueCtx) {
userRequestedStop = true
// Clear the pre-loaded next URI FIRST. Otherwise some renderers
// (specifically TVs with the AVTransport spec's "auto-play next URI
// on STOPPED transition" behavior) jump straight from the Stop
// command into playing the gapless-prefetched track. The user sees:
// "I hit Stop, the song stopped, then a different song started." We
// need to remove the receiver's idea of "what to play next" before
// we tell it to stop, so there's nothing for it to advance into.
//
// Order matters: clear-next BEFORE stop. If we did stop first, the
// receiver could trigger the next-URI before our clear-next reached
// it. Failures here are swallowed — if the clear-next fails, the
// worst case is the symptom we're trying to fix; we still try the
// Stop after.
catching("clear next") {
invokeAvtAction(
"SetNextAVTransportURI",
mapOf("NextURI" to "", "NextURIMetaData" to ""),
)
}
invokeAvtAction("Stop")
Unit
}
override suspend fun seekTo(position: Duration) {
invokeAvtAction(
"Seek",
mapOf(
"Unit" to "REL_TIME",
"Target" to formatDurationAsUpnpClock(position),
)
)
}
/** Next track in OUR queue (Android, 2026-10-29): the device only
* holds one URI, so its own Next does nothing on most renderers. Wraps to
* the start under repeat ALL. With no queue of ours (we attached to
* something already playing), the device's own Next is all there is. */
override suspend fun next(): Unit = withContext(queueCtx) {
when {
queue.isEmpty() -> { invokeAvtAction("Next"); Unit }
queueIndex < queue.lastIndex -> { sendTrackAt(queueIndex + 1, autoplay = true); Unit }
_state.value.repeatMode == RepeatMode.ALL -> { sendTrackAt(0, autoplay = true); Unit }
else -> Unit // end of the queue
}
}
/** Previous track in OUR queue, stepping back past tracks the device can't
* fetch (Android parity, 2026-10-29). On the first track it starts that track
* over; with no queue of ours it is the device's own Previous. */
override suspend fun previous(): Unit = withContext(queueCtx) {
when {
queue.isEmpty() -> { invokeAvtAction("Previous"); Unit }
queueIndex > 0 -> {
if (!sendTrackAt(queueIndex - 1, autoplay = true, step = -1)) seekTo(Duration.ZERO)
}
else -> seekTo(Duration.ZERO)
}
}
override suspend fun setVolume(volumePercent: Int) {
val v = volumePercent.coerceIn(0, 100)
// Optimistic, like the local renderer: repeated scroll notches each
// start from the value just asked for, not from a stale one that only
// an event (if any) would refresh.
_state.update { it.copy(volumePercent = v) }
invokeRcAction(
"SetVolume",
mapOf(
"Channel" to "Master",
"DesiredVolume" to v.toString(),
)
)
}
/**
* Replace the queue with [tracks] and start playing at [startIndex].
*
* The track's [DidlItem.Track.resourceUrl] must be reachable from the remote
* renderer's network position. For Earhorn-served music this Just Works
* because everyone's on the same LAN and the URL is a stable HTTP URL on
* the UPnP server.
*/
override suspend fun load(
tracks: List<DidlItem.Track>,
startIndex: Int,
autoplay: Boolean,
): Unit = withContext(queueCtx) {
queue.clear()
userRequestedStop = false
if (tracks.isEmpty()) {
queueIndex = -1
preShuffleOrder = null
publishQueueState()
return@withContext
}
queue.addAll(tracks)
queueIndex = startIndex.coerceIn(0, queue.lastIndex)
// Shuffle on: the new list is the "in order" one; only what comes after
// the start track is shuffled, as for the local renderer.
preShuffleOrder = if (_state.value.shuffleEnabled) {
queue.toList().also { shuffleTailAfterCurrent() }
} else null
// Publish the queue state to the UI BEFORE we issue the SetAVTransportURI
// network call. Without this, an outage / slow remote / wedged HTTP
// makes the user's queue silently disappear from their phone screen
// until either the call completes or the user re-navigates — they end
// up thinking "I had a queue, picked the desktop, and my queue vanished."
// Publishing first means the queue is visibly present immediately;
// sendTrackAt publishes again at the end with the post-network
// state (current-track / position) but the queue list stays consistent
// throughout.
publishQueueState()
// Android parity (2026-10-29): when nothing from startIndex on is playable
// there, the new queue is still shown (sendTrackAt publishes it).
sendTrackAt(queueIndex, autoplay)
Unit
}
// ---- Queue mutation -----------------------------------------------------
/**
* Append [tracks] to the end of the queue. Per the Renderer contract, if
* the queue was empty before this call, the tracks become the queue but
* playback is NOT auto-started — caller follows up with explicit [play].
*/
override suspend fun enqueue(tracks: List<DidlItem.Track>): Unit = withContext(queueCtx) {
if (tracks.isEmpty()) return@withContext
val wasEmpty = queue.isEmpty()
val previousLast = queue.lastIndex
queue.addAll(tracks)
if (wasEmpty) {
// Load (no autoplay) so position info / current-track-metadata
// reflects the queued track for the UI. Caller will start playback.
sendTrackAt(0, autoplay = false)
} else {
// If we just added a track that's now in the "next" slot (or, under
// repeat ALL, changed what follows the last one), pre-load it.
if (queueIndex == previousLast) preloadNextIfAny()
publishQueueState()
}
Unit
}
/**
* Insert [tracks] immediately after the currently-playing track. Per the
* Renderer contract, if the queue was empty before this call, the tracks
* become the queue AND start playing (this is the "play this album now"
* affordance, as opposed to enqueue's "play after current").
*/
override suspend fun playNext(tracks: List<DidlItem.Track>): Unit = withContext(queueCtx) {
if (tracks.isEmpty()) return@withContext
val wasEmpty = queue.isEmpty()
val insertAt = (queueIndex + 1).coerceAtLeast(0).coerceAtMost(queue.size)
queue.addAll(insertAt, tracks)
if (wasEmpty) {
sendTrackAt(0, autoplay = true)
} else {
// The track at insertAt is now the new "next". Pre-load it.
preloadNextIfAny()
publishQueueState()
}
Unit
}
/** Jump to a specific queue position. Plays immediately on success. */
override suspend fun skipToQueueItem(index: Int): Unit = withContext(queueCtx) {
if (index < 0 || index >= queue.size) return@withContext
sendTrackAt(index, autoplay = true)
Unit
}
/**
* Remove a track at [index] from the queue. Behavior matches the local
* renderer's contract:
* - Removing the currently-playing track advances to the next; if there
* is no next, playback stops.
* - Removing a track before the current adjusts [queueIndex] so the
* same track stays "current".
* - Removing a track after the current is invisible to playback but
* may invalidate the pre-loaded next.
*/
override suspend fun removeFromQueue(index: Int): Unit = withContext(queueCtx) {
if (index < 0 || index >= queue.size) return@withContext
val wasCurrent = index == queueIndex
val wasBeforeCurrent = index < queueIndex
queue.removeAt(index)
if (wasCurrent) {
// After removeAt(queueIndex), what was queue[queueIndex+1] is now
// at queue[queueIndex]. If queueIndex is still in range, that's
// our new "current". If it's out of range, we removed the last
// item in the queue — stop, don't loop back.
if (queueIndex >= queue.size) {
queueIndex = -1
stop()
publishQueueState()
} else {
sendTrackAt(queueIndex, autoplay = _state.value.isPlaying)
}
} else {
if (wasBeforeCurrent) queueIndex -= 1
// The pre-loaded next may now be wrong; re-issue.
preloadNextIfAny()
publishQueueState()
}
Unit
}
/**
* Move a track from [from] to [to]. Adjusts [queueIndex] so the currently-
* playing track stays "current" even if reorder shifts its position.
*/
override suspend fun moveInQueue(from: Int, to: Int): Unit = withContext(queueCtx) {
if (from == to) return@withContext
if (from !in queue.indices || to !in queue.indices) return@withContext
val item = queue.removeAt(from)
queue.add(to, item)
// Track the queueIndex carefully through the shift.
queueIndex = when {
from == queueIndex -> to
from < queueIndex && to >= queueIndex -> queueIndex - 1
from > queueIndex && to <= queueIndex -> queueIndex + 1
else -> queueIndex
}
preloadNextIfAny()
publishQueueState()
}
/** Empty the queue and stop playback. */
override suspend fun clearQueue(): Unit = withContext(queueCtx) {
queue.clear()
queueIndex = -1
preShuffleOrder = null
stop()
// publishQueueState keeps the old `current` when the queue is empty
// (it may be the device's own track); after a clear there is none.
_state.update { it.copy(queue = emptyList(), queueIndex = -1, current = null) }
}
// ---- Queue helpers ------------------------------------------------------
/** Android parity (2026-10-29): true when the remote device can fetch [track].
* Local files (file://, plain paths) only exist on this computer. */
private fun isRemotePlayable(track: DidlItem.Track): Boolean =
track.resourceUrl.startsWith("http://", ignoreCase = true) ||
track.resourceUrl.startsWith("https://", ignoreCase = true)
/**
* Send `queue[index]` to the remote via SetAVTransportURI, optionally start
* it, then pre-load the next track for gapless transitions. Tracks the
* device can't fetch are skipped, moving by [step] (Android,
* 2026-10-29); when there is none, a forward move stops. Returns true when
* a track was sent. Records the URI in [pendingExpectedUri] so its echo
* isn't read as a change of track.
*/
private suspend fun sendTrackAt(index: Int, autoplay: Boolean, step: Int = 1): Boolean {
lastTrackSentAt = System.currentTimeMillis()
var playable = index
while (playable in queue.indices && !isRemotePlayable(queue[playable])) playable += step
val track = queue.getOrNull(playable) ?: run {
log("No http(s) track from index $index on — local files can't play on ${target.friendlyName}")
if (step > 0 && index in queue.indices) stop()
publishQueueState()
return false
}
if (playable != index) log("Skipped ${kotlin.math.abs(playable - index)} local-only track(s)")
queueIndex = playable
val didl = DidlLiteCodec.encodeTrack(track)
previousRemoteUri = currentRemoteUri
previousRemoteAlias = currentRemoteAlias
currentRemoteUri = track.resourceUrl
currentRemoteAlias = null
pendingExpectedUri = track.resourceUrl
lastSeenPosition = Duration.ZERO
// The new track's own length, not the previous one's: the end-of-track
// test compares against it. Events / the poller refine it.
_state.update {
it.copy(
position = Duration.ZERO,
duration = track.duration?.takeIf { d -> d > Duration.ZERO } ?: Duration.ZERO,
)
}
publishQueueState()
invokeAvtAction(
"SetAVTransportURI",
mapOf(
"CurrentURI" to track.resourceUrl,
"CurrentURIMetaData" to didl,
)
)
if (autoplay) play()
publishQueueState()
preloadNextIfAny()
return true
}
/**
* Tell the remote the URI of the next track in our queue via
* SetNextAVTransportURI. Most renderers (Linn, BubbleUPnP, gmediarender,
* Earhorn) honor this for gapless playback — they transition from the
* current track to the next without an audible pause.
*
* Pass an empty URI to clear any previously-set next track. We do this
* when our queue runs out so the renderer knows to stop rather than
* loop back to a stale next-URI.
*
* Failures are logged and swallowed — pre-load is a nice-to-have, not
* a correctness requirement. The auto-advance detection in
* [handleCurrentTrackUriChange] handles the case where pre-load failed
* by still recognizing our queue ownership and advancing locally.
*/
private suspend fun preloadNextIfAny() {
// Repeat-aware, so repeat ALL wraps gaplessly. Repeat ONE is the
// device's own REPEAT_ONE (see remotePlayMode); a track it can't fetch
// is left to the STOPPED fallback, which skips it.
val next = nextAutoIndex()
?.takeIf { _state.value.repeatMode != RepeatMode.ONE }
?.let(queue::getOrNull)
?.takeIf(::isRemotePlayable)
if (next == null) {
// Clear any previously-set next URI so the renderer doesn't
// auto-advance to stale state when our current track ends.
invokeAvtAction(
"SetNextAVTransportURI",
mapOf("NextURI" to "", "NextURIMetaData" to "")
)
return
}
val didl = DidlLiteCodec.encodeTrack(next)
invokeAvtAction(
"SetNextAVTransportURI",
mapOf(
"NextURI" to next.resourceUrl,
"NextURIMetaData" to didl,
)
)
}
/**
* Snapshot the current queue state into [_state]. Called after every
* queue mutation. Preserves any non-queue fields (transport state,
* volume, position) that came from GENA events on the remote.
*/
private fun publishQueueState() {
_state.value = _state.value.copy(
queue = queue.toList(),
queueIndex = queueIndex,
current = queue.getOrNull(queueIndex) ?: _state.value.current,
)
}
/**
* Called for every `CurrentTrackURI` the remote reports. Renderers resend it
* with every LastChange (Earhorn's own renderer does), so most calls are
* not a change at all. In order:
*
* 1. **Our own SetAVTransportURI echoed back** (matches [pendingExpectedUri]).
* 2. **The current track again** (or what the device calls it), or a late
* echo of the track we just replaced — nothing to do. These used to fall
* through to case 5 and wipe the queue (Team 3 P1-8).
* 3. **A rewritten form of the track we just sent** (a proxy or redirect),
* reported within [SEND_GRACE_MS] of sending it: remember it as that
* track's alias.
* 4. **The device moved within our queue** — normally the gapless switch to
* the pre-loaded next track ([nextAutoIndex]), or any other queue entry.
* 5. **External hijack** (matches nothing in our queue) — another control
* point sent SetAVTransportURI and changed what's playing. Our queue is
* stale; let it go and show the device's own metadata.
*/
private suspend fun handleCurrentTrackUriChange(newUri: String) {
if (newUri.isBlank() || newUri == "about:blank" || newUri == "NOT_IMPLEMENTED") return
// Case 1: we set this URI ourselves.
if (pendingExpectedUri == newUri) {
pendingExpectedUri = null
return
}
// Case 2: not a change.
if (newUri == queue.getOrNull(queueIndex)?.resourceUrl || newUri == currentRemoteAlias) return
if (newUri == previousRemoteUri || newUri == previousRemoteAlias) return
// Case 3: our just-sent track under another name.
if (pendingExpectedUri != null &&
System.currentTimeMillis() - lastTrackSentAt < SEND_GRACE_MS
) {
log("Renderer reports a rewritten URI for the track just sent; treating it as that track")
currentRemoteAlias = newUri
pendingExpectedUri = null
return
}
// Case 4: the device moved within our queue — the pre-loaded next
// track first (gapless, repeat-aware), then any other entry.
val nextIdx = nextAutoIndex()
val idx = if (nextIdx != null && queue[nextIdx].resourceUrl == newUri) nextIdx
else queue.indexOfFirst { it.resourceUrl == newUri }
if (idx >= 0) {
queueIndex = idx
previousRemoteUri = currentRemoteUri
previousRemoteAlias = currentRemoteAlias
currentRemoteUri = newUri
currentRemoteAlias = null
pendingExpectedUri = null
// A STOPPED blip right after the switch must not advance again.
lastTrackSentAt = System.currentTimeMillis()
lastSeenPosition = Duration.ZERO
_state.update {
it.copy(
position = Duration.ZERO,
duration = queue[idx].duration?.takeIf { d -> d > Duration.ZERO } ?: Duration.ZERO,
)
}
publishQueueState()
preloadNextIfAny()
return
}
// Case 5: external hijack. Release queue ownership.
if (queue.isNotEmpty()) {
log("CurrentTrackURI doesn't match our queue; another controller took over — releasing the queue")
queue.clear()
queueIndex = -1
preShuffleOrder = null
currentRemoteUri = null
currentRemoteAlias = null
publishQueueState()
}
}
// ---- Auto-advance (STOPPED fallback) ------------------------------------
/** Android parity (2026-10-29): where playback goes when OUR track ended —
* repeat ONE replays it, repeat ALL wraps to the start, otherwise the
* next track or nothing. */
private fun nextAutoIndex(): Int? = when {
queue.isEmpty() || queueIndex !in queue.indices -> null
_state.value.repeatMode == RepeatMode.ONE -> queueIndex
queueIndex < queue.lastIndex -> queueIndex + 1
_state.value.repeatMode == RepeatMode.ALL -> 0
else -> null
}
/**
* The device went STOPPED (seen by an event or by the poller). Android's
* rule 2026-09-02: advance only when OUR track played to its end — the last
* position seen is within 3 s of its length. A STOPPED right after we sent a
* track is the device switching tracks and is ignored; any other stop (the
* speaker's buttons, another app) is a stop, and nothing is sent.
*/
private suspend fun onRemoteStopped() {
if (userRequestedStop) return
if (System.currentTimeMillis() - lastTrackSentAt <= SEND_GRACE_MS) return
val dur = _state.value.duration
val reachedEnd = dur > Duration.ZERO && dur - lastSeenPosition <= END_OF_TRACK_SLACK
val next = nextAutoIndex()
if (!reachedEnd || next == null) {
userRequestedStop = true // a stop from anywhere is a stop
return
}
log("Our track ended — sending queue index $next")
sendTrackAt(next, autoplay = true)
}
// ---- Repeat / shuffle ---------------------------------------------------
override suspend fun setRepeatMode(mode: RepeatMode): Unit = withContext(queueCtx) {
// Android parity (2026-10-29): keep the mode in our state (the UI cycles from
// it; it was never set, so every press sent ALL). Repeat is applied by
// our own advance logic (nextAutoIndex). See remotePlayMode for what
// the device itself is told.
_state.update { it.copy(repeatMode = mode) }
invokeAvtAction("SetPlayMode", mapOf("NewPlayMode" to remotePlayMode()))
// Repeat changes what "next" is at the end of the queue.
if (queue.isNotEmpty()) preloadNextIfAny()
Unit
}
/**
* Keep the flag in our state so the toggle can be turned off again
* (Android parity, 2026-10-29), and shuffle OUR queue: the tracks after the
* current one, as the local renderer does. Turning it off restores the
* earlier order, keeping tracks added meanwhile (at the end) and leaving out
* tracks removed meanwhile.
*/
override suspend fun setShuffle(enabled: Boolean): Unit = withContext(queueCtx) {
if (enabled != _state.value.shuffleEnabled && queue.isNotEmpty()) {
if (enabled) {
preShuffleOrder = queue.toList()
shuffleTailAfterCurrent()
} else {
preShuffleOrder?.let { restoreOrder(it) }
preShuffleOrder = null
}
}
if (!enabled) preShuffleOrder = null
_state.update { it.copy(shuffleEnabled = enabled) }
invokeAvtAction("SetPlayMode", mapOf("NewPlayMode" to remotePlayMode()))
if (queue.isNotEmpty()) {
publishQueueState()
preloadNextIfAny()
}
Unit
}
private fun shuffleTailAfterCurrent() {
val from = (queueIndex + 1).coerceIn(0, queue.size)
if (queue.size - from < 2) return
val tail = queue.subList(from, queue.size).shuffled()
while (queue.size > from) queue.removeAt(queue.lastIndex)
queue.addAll(tail)
}
/** Put [queue] back in [order]'s sequence. Entries are matched as the same
* objects, so duplicates and the current track keep their identity. */
private fun restoreOrder(order: List<DidlItem.Track>) {
val current = queue.getOrNull(queueIndex)
val remaining = queue.toMutableList()
val restored = ArrayList<DidlItem.Track>(queue.size)
for (t in order) {
val i = remaining.indexOfFirst { it === t }
if (i >= 0) restored += remaining.removeAt(i)
}
restored += remaining // added while shuffled
queue.clear()
queue.addAll(restored)
if (current != null) queueIndex = queue.indexOfFirst { it === current }.coerceAtLeast(0)
}
/** Android parity (2026-10-29): the PlayMode the device itself gets. It only ever
* holds ONE track (we send them one at a time), so REPEAT_ALL there just
* loops that track and our queue never advances — repeat ALL is ours to
* do, and the device stays NORMAL. REPEAT_ONE on the device is the same
* thing we'd do. */
private fun remotePlayMode(): String {
val s = _state.value
return when {
s.shuffleEnabled -> "SHUFFLE"
s.repeatMode == RepeatMode.ONE -> "REPEAT_ONE"
else -> "NORMAL"
}
}
// ---- Initial state pulls ------------------------------------------------
private suspend fun pullInitialAvtState(device: RemoteDevice) {
val info = invokeAvtAction(device, "GetTransportInfo", emptyMap())
val transportStateStr = info?.get("CurrentTransportState") ?: return
val media = invokeAvtAction(device, "GetMediaInfo", emptyMap())
val mediaUri = media?.get("CurrentURI").orEmpty()
val mediaDuration = media?.get("MediaDuration").orEmpty()
val mediaMeta = media?.get("CurrentURIMetaData").orEmpty()
val track = DidlLiteCodec.decodeFirstTrack(mediaMeta, mediaUri, requirePlayableUrl = false)
_state.update {
it.copy(
// Our own queue (a load that raced ahead of this pull) wins.
current = if (queue.isEmpty()) track ?: it.current else it.current,
isPlaying = transportStateStr == "PLAYING",
isStopped = transportStateStr == "STOPPED" || transportStateStr == "NO_MEDIA_PRESENT",
isBuffering = transportStateStr == "TRANSITIONING",
duration = if (queue.isEmpty()) parseHmsToDuration(mediaDuration) ?: Duration.ZERO else it.duration,
)
}
}
private suspend fun pullInitialVolume(device: RemoteDevice) {
val out = invokeRcAction(device, "GetVolume", mapOf("Channel" to "Master"))
val pct = out?.get("CurrentVolume")?.toIntOrNull() ?: return
_state.update { it.copy(volumePercent = pct.coerceIn(0, 100)) }
}
// ---- LastChange parsing -------------------------------------------------
private val avtParser by lazy { AVTransportLastChangeParser() }
private val rcParser by lazy { RenderingControlLastChangeParser() }
/** One AVTransport LastChange, on [queueCtx]: mirror it into [state], then
* act on a reported track URI and on a fresh STOPPED. */
private suspend fun onAvtLastChange(xml: String) {
val wasStopped = _state.value.isStopped
val uri = parseAvtLastChange(xml)
if (uri != null) handleCurrentTrackUriChange(uri)
if (_state.value.isStopped && !wasStopped) onRemoteStopped()
}
/** Applies the event to [_state]; returns the CurrentTrackURI it carried. */
private fun parseAvtLastChange(xml: String): String? {
var dispatchUri: String? = null
runCatching {
// LastChangeParser.parse(xml) returns an Event with one or more
// InstanceID entries. Each InstanceID holds the EventedValue list
// for that instance. We're single-instance (ID=0) and ignore the rest.
val event = avtParser.parse(xml) ?: return@runCatching
val instance = event.instanceIDs?.firstOrNull() ?: return@runCatching
val values = instance.values ?: return@runCatching
var cur = _state.value
values.forEach { v: EventedValue<*> ->
when (v) {
is AVTransportVariable.TransportState -> {
val s = v.value
cur = cur.copy(
isPlaying = s == org.jupnp.support.model.TransportState.PLAYING,
isStopped = s == org.jupnp.support.model.TransportState.STOPPED ||
s == org.jupnp.support.model.TransportState.NO_MEDIA_PRESENT,
isBuffering = s == org.jupnp.support.model.TransportState.TRANSITIONING,
)
}
is AVTransportVariable.CurrentTrackMetaData -> {
// Only consume the remote's echoed metadata when we DON'T
// own the queue. When we do, queue[queueIndex] is the
// source of truth for `current` and the metadata echo
// would just round-trip our own data back to us — or
// worse, overwrite a richer local Track with a sparser
// remote re-serialization.
if (queue.isEmpty()) {
val didl = v.value
val uriHint = cur.current?.resourceUrl.orEmpty()
DidlLiteCodec.decodeFirstTrack(didl, uriHint, requirePlayableUrl = false)?.let {
cur = cur.copy(current = it)
}
}
}
is AVTransportVariable.CurrentTrackDuration -> {
parseHmsToDuration(v.value)?.let { cur = cur.copy(duration = it) }
}
is AVTransportVariable.CurrentTrackURI -> {
// Defer the work to handleCurrentTrackUriChange (suspend).
// The whole point of this branch is queue auto-advance
// detection: when the renderer transitions from current
// to the pre-loaded next, this fires with the new URI.
val u = v.value?.toString()?.takeIf { it.isNotBlank() }
if (u != null) dispatchUri = u
}
// TransportStatus, AVTransportURI, AVTransportURIMetaData,
// CurrentMediaDuration, NumberOfTracks, CurrentTrack — not
// mirrored separately; CurrentTrackMetaData covers the
// display surface and the queue covers the rest.
else -> { /* ignore */ }
}
}
_state.value = cur
}.onFailure {
log("parseAvtLastChange threw: ${it.message} — attempting regex fallback")
dispatchUri = runCatching { parseAvtLastChangeFallback(xml) }
.onFailure { fb -> log("regex fallback also failed: ${fb.message}") }
.getOrNull()
}
return dispatchUri
}
/**
* Defensive regex-based fallback for [parseAvtLastChange] when jUPnP's
* [AVTransportLastChangeParser] throws — typically because the support
* XSDs aren't on the classpath (a known issue on Android where the
* `org/fourthline/cling/support/` resources are missing, and a
* theoretical issue on desktop if a future jUPnP shading pass drops
* them). Rather than lose state updates entirely on the failure
* branch, we extract the variables we care about with a handful of
* regexes and apply the same `_state` mutations the proper parser
* would have.
*
* Patterns are deliberately namespace-tolerant — some renderers
* prefix AVT variables (`<avt:TransportState val="…"/>`) and our
* regex needs to match either form. We also accept the val attribute
* in any position within the element rather than insisting it come
* first.
*
* The CurrentTrackMetaData value arrives HTML-entity-encoded (it's
* XML embedded in XML, escaped to keep parsers happy), so we run a
* minimal entity decode before handing the DIDL-Lite payload to
* [DidlLiteCodec.decodeFirstTrack]. The decode order matters: `&`
* must be replaced last so we don't double-decode entities like
* `&lt;` into `<` instead of `<`.
*/
private fun parseAvtLastChangeFallback(xml: String): String? {
val transportStateRx = Regex(
"""<(?:[\w-]+:)?TransportState[^>]*\bval="([^"]+)""""
)
val durationRx = Regex(
"""<(?:[\w-]+:)?CurrentTrackDuration[^>]*\bval="([^"]+)""""
)
val metadataRx = Regex(
"""<(?:[\w-]+:)?CurrentTrackMetaData[^>]*\bval="([^"]*)""""
)
val uriRx = Regex(
"""<(?:[\w-]+:)?CurrentTrackURI[^>]*\bval="([^"]+)""""
)
var cur = _state.value
var dispatchUri: String? = null
transportStateRx.find(xml)?.groupValues?.getOrNull(1)?.let { s ->
if (s == "PLAYING" && !cur.isPlaying) userRequestedStop = false // see the poller
cur = cur.copy(
isPlaying = s == "PLAYING",
isStopped = s == "STOPPED" || s == "NO_MEDIA_PRESENT",
isBuffering = s == "TRANSITIONING",
)
}
durationRx.find(xml)?.groupValues?.getOrNull(1)?.let { hms ->
parseHmsToDuration(hms)?.let { cur = cur.copy(duration = it) }
}
if (queue.isEmpty()) {
metadataRx.find(xml)?.groupValues?.getOrNull(1)?.let { encoded ->
val didl = decodeHtmlEntities(encoded)
val uriHint = cur.current?.resourceUrl.orEmpty()
DidlLiteCodec.decodeFirstTrack(didl, uriHint, requirePlayableUrl = false)?.let {
cur = cur.copy(current = it)
}
}
}
uriRx.find(xml)?.groupValues?.getOrNull(1)
?.takeIf { it.isNotBlank() }
?.let { dispatchUri = it }
_state.value = cur
return dispatchUri
}
/**
* Minimal HTML entity decoder for the five entities that show up in
* UPnP-escaped DIDL-Lite payloads. `&` must be replaced LAST or
* we'll double-decode (`&lt;` would become `<` instead of the
* intended `<`).
*/
private fun decodeHtmlEntities(s: String): String =
s.replace("<", "<")
.replace(">", ">")
.replace(""", "\"")
.replace("'", "'")
.replace("&", "&")
private fun parseRcLastChange(xml: String) {
runCatching {
val event = rcParser.parse(xml) ?: return@runCatching
val instance = event.instanceIDs?.firstOrNull() ?: return@runCatching
val values = instance.values ?: return@runCatching
var cur = _state.value
values.forEach { v: EventedValue<*> ->
when (v) {
is RenderingControlVariable.Volume -> {
// ChannelVolume getters: channel + volume (Integer).
val cv = v.value
if (cv?.channel == org.jupnp.support.model.Channel.Master) {
cv.volume?.toInt()?.coerceIn(0, 100)?.let { vol ->
cur = cur.copy(volumePercent = vol)
}
}
}
is RenderingControlVariable.Mute -> {
// We don't track mute in PlaybackState; volume = 0 conveys it.
}
else -> { /* Loudness, PresetNameList, etc. — ignored */ }
}
}
_state.value = cur
}.onFailure { log("parseRcLastChange threw: ${it.message}") }
}
// ---- Poller -------------------------------------------------------------
/** Poll GetTransportInfo (1 s while playing, 10 s otherwise) and
* GetPositionInfo while playing. This is also the auto-advance path when
* GENA events never arrive (Android's poller, Android parity
* 2026-09-02: still notices someone pressing Play on the speaker itself,
* without sixty requests a minute for nothing). */
private fun startPositionPoller() {
positionPollerJob?.cancel()
positionPollerJob = scope.launch(queueCtx) {
while (true) {
val device = findRemoteDevice()
if (device != null) {
catching("GetTransportInfo") {
val ts = invokeAvtAction(device, "GetTransportInfo", emptyMap())
?.get("CurrentTransportState")
if (!ts.isNullOrBlank()) {
val wasStopped = _state.value.isStopped
val nowStopped = ts == "STOPPED" || ts == "NO_MEDIA_PRESENT"
// Playing again after a stop from anywhere (e.g. Play
// on the speaker itself): the queue should carry on
// at the end of this track, so re-arm auto-advance.
if (ts == "PLAYING" && !_state.value.isPlaying) userRequestedStop = false
_state.update {
it.copy(
isPlaying = ts == "PLAYING",
isStopped = nowStopped,
isBuffering = ts == "TRANSITIONING",
)
}
if (nowStopped && !wasStopped) onRemoteStopped()
}
}
// Position while playing; remembered for the end-of-track test.
if (_state.value.isPlaying) {
catching("GetPositionInfo") {
val out = invokeAvtAction(device, "GetPositionInfo", emptyMap())
// A renderer that honours SetNextAVTransportURI but
// sends no events moves to the pre-loaded track with
// no STOPPED we'd see. Its TrackURI is the only sign;
// without this the UI keeps the old track and the
// end-of-track test runs against the wrong length.
// Only moves WITHIN our queue are acted on here; a URI
// that matches nothing could be the device's rewritten
// name for our track, and treating that as a takeover
// from a poll would drop the queue. Takeovers are left
// to the evented path.
out?.get("TrackURI")
?.takeIf { uri -> uri.isNotBlank() && queue.any { it.resourceUrl == uri } }
?.let { uri -> handleCurrentTrackUriChange(uri) }
parseHmsToDuration(out?.get("RelTime"))?.let { p ->
_state.update { it.copy(position = p) }
lastSeenPosition = p
}
if (_state.value.duration == Duration.ZERO) {
parseHmsToDuration(out?.get("TrackDuration"))?.let { d ->
_state.update { it.copy(duration = d) }
}
}
}
}
}
delay(if (_state.value.isPlaying) 1_000 else 10_000)
}
}
}
// ---- jUPnP action plumbing ----------------------------------------------
/** Invoke a named action on the remote device's AVTransport service. Returns the
* output args by name, or null on failure. Suspends until the action completes. */
private suspend fun invokeAvtAction(
actionName: String,
args: Map<String, String> = emptyMap(),
): Map<String, String>? {
val device = findRemoteDevice() ?: return null
return invokeAvtAction(device, actionName, args)
}
private suspend fun invokeAvtAction(
device: RemoteDevice,
actionName: String,
args: Map<String, String>,
): Map<String, String>? {
val service = device.findAvtService() ?: return null
return invokeOnService(service, actionName, args)
}
private suspend fun invokeRcAction(
actionName: String,
args: Map<String, String> = emptyMap(),
): Map<String, String>? {
val device = findRemoteDevice() ?: return null
return invokeRcAction(device, actionName, args)
}
private suspend fun invokeRcAction(
device: RemoteDevice,
actionName: String,
args: Map<String, String>,
): Map<String, String>? {
val service = device.findRcService() ?: return null
return invokeOnService(service, actionName, args)
}
/** Core suspend-wrap of jUPnP's callback-based execute(). Always includes
* `InstanceID=0` as the first argument since Earhorn is single-instance. */
private suspend fun invokeOnService(
service: RemoteService,
actionName: String,
args: Map<String, String>,
): Map<String, String>? {
val action = service.getAction(actionName) ?: run {
log("Service has no '$actionName' action")
return null
}
val controlPoint = holder.controlPoint ?: return null
return suspendCancellableCoroutine { cont ->
val invocation = ActionInvocation(action)
// InstanceID is the conventional first arg for both AVT and RC.
// Renderers that don't support multi-instance ignore non-zero IDs.
runCatching { invocation.setInput("InstanceID", "0") }
args.forEach { (k, v) ->
runCatching { invocation.setInput(k, v) }
}
val callback = object : ActionCallback(invocation) {
override fun success(inv: ActionInvocation<*>?) {
val outputs = inv?.toOutputMap().orEmpty()
if (cont.isActive) cont.resume(outputs)
}
override fun failure(
inv: ActionInvocation<*>?,
operation: UpnpResponse?,
defaultMsg: String?,
) {
val cause = inv?.failure as? ActionException ?: ActionException(
org.jupnp.model.types.ErrorCode.ACTION_FAILED,
defaultMsg ?: "Unknown error"
)
log("Action $actionName failed: ${cause.message}")
if (cont.isActive) cont.resume(null)
}
}
controlPoint.execute(callback)
}
}
/** Convert an [ActionInvocation]'s output arguments into a name→string-value map.
* Function (not property) to avoid colliding with whatever name jUPnP may already
* expose on the class — extension properties don't override member properties,
* and the compiler errors get confusing fast. */
private fun ActionInvocation<*>.toOutputMap(): Map<String, String> =
output.orEmpty().associate { it.argument.name to (it.value?.toString().orEmpty()) }
// ---- Helpers ------------------------------------------------------------
private fun findRemoteDevice(): RemoteDevice? = holder.findDevice(target.udn)
private fun RemoteDevice.findAvtService(): RemoteService? =
findService(UDAServiceType("AVTransport", 1)) as? RemoteService
private fun RemoteDevice.findRcService(): RemoteService? =
findService(UDAServiceType("RenderingControl", 1)) as? RemoteService
private fun formatDurationAsUpnpClock(duration: Duration): String {
val totalMs = duration.inWholeMilliseconds.coerceAtLeast(0)
val ms = totalMs % 1000
val total = totalMs / 1000
val h = total / 3600
val m = (total % 3600) / 60
val s = total % 60
return "%d:%02d:%02d.%03d".format(h, m, s, ms)
}
private fun parseHmsToDuration(clock: String?): Duration? {
if (clock.isNullOrBlank() || clock == "NOT_IMPLEMENTED") return null
val parts = clock.split(":")
if (parts.size != 3) return null
val h = parts[0].toLongOrNull() ?: return null
val m = parts[1].toLongOrNull() ?: return null
val s = parts[2].toDoubleOrNull() ?: return null
val ms = ((h * 3600 + m * 60) * 1000 + (s * 1000).toLong())
return ms.milliseconds
}
/** try/catch for the network steps above: logs and returns null on
* failure, but lets cancellation through (runCatching would swallow it). */
private inline fun <T> catching(what: String, block: () -> T): T? = try {
block()
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
log("$what failed: ${e.message}")
null
}
private fun log(msg: String) {
println("[RemoteUpnpRenderer/${target.friendlyName}] $msg")
}
private companion object {
/** After sending a track: no STOPPED counts as "ended", and a different
* URI counts as the device's name for it (Android parity, 2026-09-02). */
const val SEND_GRACE_MS = 5_000L
/** "Reached the end": last position seen within this of the length. */
val END_OF_TRACK_SLACK = 3.seconds
}
}
