mirror of
https://github.com/KRTirtho/spotube.git
synced 2026-09-20 06:34:00 +00:00
feat(remote-control): refactor service registration and cleanup logic for improved reliability
This commit is contained in:
parent
9ec7704d7b
commit
571aea8d38
@ -235,7 +235,12 @@ actual class AudioPlayer actual constructor(context: Any) : AudioPlayerInterface
|
|||||||
|
|
||||||
actual override suspend fun seekTo(position: Duration) {
|
actual override suspend fun seekTo(position: Duration) {
|
||||||
withContext(Dispatchers.Main) {
|
withContext(Dispatchers.Main) {
|
||||||
val targetMs = position.inWholeMilliseconds.coerceIn(0, exoPlayer.duration)
|
val duration = exoPlayer.duration
|
||||||
|
val targetMs = if (duration > 0) {
|
||||||
|
position.inWholeMilliseconds.coerceIn(0, duration)
|
||||||
|
} else {
|
||||||
|
position.inWholeMilliseconds.coerceAtLeast(0)
|
||||||
|
}
|
||||||
exoPlayer.seekTo(targetMs)
|
exoPlayer.seekTo(targetMs)
|
||||||
_position.tryEmit(exoPlayer.currentPosition.milliseconds)
|
_position.tryEmit(exoPlayer.currentPosition.milliseconds)
|
||||||
}
|
}
|
||||||
|
|||||||
@ -90,13 +90,19 @@ class DeviceDiscoveryService {
|
|||||||
deviceId: String,
|
deviceId: String,
|
||||||
registerTimeoutMs: Long = 5_000,
|
registerTimeoutMs: Long = 5_000,
|
||||||
): NetService {
|
): NetService {
|
||||||
val service = createNetService(
|
val service = createService(name, port, deviceId)
|
||||||
type = SERVICE_TYPE,
|
|
||||||
name = name,
|
|
||||||
port = port,
|
|
||||||
txt = mapOf(TXT_DEVICE_ID to deviceId),
|
|
||||||
)
|
|
||||||
service.register(timeoutInMs = registerTimeoutMs)
|
service.register(timeoutInMs = registerTimeoutMs)
|
||||||
return service
|
return service
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fun createService(
|
||||||
|
name: String,
|
||||||
|
port: Int,
|
||||||
|
deviceId: String,
|
||||||
|
): NetService = createNetService(
|
||||||
|
type = SERVICE_TYPE,
|
||||||
|
name = name,
|
||||||
|
port = port,
|
||||||
|
txt = mapOf(TXT_DEVICE_ID to deviceId),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
@ -73,7 +73,6 @@ class JamSessionService(
|
|||||||
audioPlayer = audioPlayer,
|
audioPlayer = audioPlayer,
|
||||||
audioPlayerQueue = audioPlayerQueue,
|
audioPlayerQueue = audioPlayerQueue,
|
||||||
jamSession = this,
|
jamSession = this,
|
||||||
settingsProvider = settingsProvider,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
private val json = Json {
|
private val json = Json {
|
||||||
|
|||||||
@ -23,10 +23,8 @@ import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.artist.MetadataArtist
|
|||||||
import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.track.MetadataTrack
|
import dev.krtirtho.plugin_interfaces.plugin_apis.metadata.track.MetadataTrack
|
||||||
import dev.krtirtho.spotube.core.audioplayer.AudioPlayerInterface
|
import dev.krtirtho.spotube.core.audioplayer.AudioPlayerInterface
|
||||||
import dev.krtirtho.spotube.core.audioplayer.AudioPlayerQueue
|
import dev.krtirtho.spotube.core.audioplayer.AudioPlayerQueue
|
||||||
import dev.krtirtho.spotube.core.audioplayer.MediaItem
|
|
||||||
import dev.krtirtho.spotube.core.audioplayer.PlayerState
|
import dev.krtirtho.spotube.core.audioplayer.PlayerState
|
||||||
import dev.krtirtho.spotube.core.audioplayer.QueueEntry
|
import dev.krtirtho.spotube.core.audioplayer.QueueEntry
|
||||||
import dev.krtirtho.spotube.modules.settings.SettingsProvider
|
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
import kotlinx.coroutines.Job
|
import kotlinx.coroutines.Job
|
||||||
@ -54,7 +52,6 @@ class QueueSyncManager(
|
|||||||
private val audioPlayer: AudioPlayerInterface,
|
private val audioPlayer: AudioPlayerInterface,
|
||||||
private val audioPlayerQueue: AudioPlayerQueue,
|
private val audioPlayerQueue: AudioPlayerQueue,
|
||||||
private val jamSession: JamSessionService,
|
private val jamSession: JamSessionService,
|
||||||
private val settingsProvider: SettingsProvider,
|
|
||||||
) {
|
) {
|
||||||
private val log = Logger.withTag("QueueSyncManager")
|
private val log = Logger.withTag("QueueSyncManager")
|
||||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||||
@ -198,12 +195,15 @@ class QueueSyncManager(
|
|||||||
if (queueChanged) {
|
if (queueChanged) {
|
||||||
lastAppliedItems = playableItems
|
lastAppliedItems = playableItems
|
||||||
lastAppliedCurrentIndex = state.currentIndex
|
lastAppliedCurrentIndex = state.currentIndex
|
||||||
val mediaItems = playableItems.map { it.toPlayableMediaItem() }
|
val entries = playableItems.map { it.toQueueEntry() }
|
||||||
runCatching {
|
runCatching {
|
||||||
audioPlayer.load(
|
// Load through the queue repository (like the host does) so the
|
||||||
playlist = mediaItems,
|
// stream proxy can resolve the tracks — it only knows tracks in
|
||||||
|
// queueFlow.
|
||||||
|
audioPlayerQueue.load(
|
||||||
|
entries = entries,
|
||||||
autoPlay = state.isPlaying,
|
autoPlay = state.isPlaying,
|
||||||
startPosition = state.currentIndex.coerceIn(0, mediaItems.lastIndex.coerceAtLeast(0)),
|
startPosition = state.currentIndex.coerceIn(0, entries.lastIndex.coerceAtLeast(0)),
|
||||||
)
|
)
|
||||||
}.onFailure { e ->
|
}.onFailure { e ->
|
||||||
log.e(e) { "Failed to apply jam queue to local player" }
|
log.e(e) { "Failed to apply jam queue to local player" }
|
||||||
@ -260,9 +260,10 @@ class QueueSyncManager(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------- Conversions ----------
|
/**
|
||||||
|
* Build a queue entry from a jam media item. Streaming tracks carry their id
|
||||||
/** Host side: turn a suggested item into a playable queue entry. */
|
* so the device's own queue/stream proxy can resolve a playable URL later.
|
||||||
|
*/
|
||||||
private fun JamMediaItem.toQueueEntry(): QueueEntry = when {
|
private fun JamMediaItem.toQueueEntry(): QueueEntry = when {
|
||||||
trackId.isNotBlank() -> QueueEntry.StreamingTrack(
|
trackId.isNotBlank() -> QueueEntry.StreamingTrack(
|
||||||
track = MetadataTrack(
|
track = MetadataTrack(
|
||||||
@ -296,41 +297,6 @@ class QueueSyncManager(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Guest side: build a playable MediaItem. Streaming tracks have their stream
|
|
||||||
* URL resolved through this device's own playback proxy (the host never sends
|
|
||||||
* usable URLs — each guest must fetch from its own plugins).
|
|
||||||
*/
|
|
||||||
private suspend fun JamMediaItem.toPlayableMediaItem(): MediaItem {
|
|
||||||
if (trackId.isNotBlank()) {
|
|
||||||
val proxyUrl = buildStreamingUrl(trackId, protocol)
|
|
||||||
return MediaItem(
|
|
||||||
title = title,
|
|
||||||
artist = artist,
|
|
||||||
album = album,
|
|
||||||
duration = kotlin.time.Duration.parse("${durationMs}ms"),
|
|
||||||
coverURL = coverUrl,
|
|
||||||
url = proxyUrl,
|
|
||||||
protocol = runCatching { StreamProtocol.valueOf(protocol.ifBlank { "PROGRESSIVE" }) }
|
|
||||||
.getOrDefault(StreamProtocol.PROGRESSIVE),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
return JamMediaItem.toMediaItem(this)
|
|
||||||
}
|
|
||||||
|
|
||||||
private suspend fun buildStreamingUrl(trackId: String, protocol: String): String {
|
|
||||||
val port = settingsProvider.settingsState
|
|
||||||
.first()
|
|
||||||
?.playbackProxyServerPort ?: return ""
|
|
||||||
val baseUrl = "http://127.0.0.1:$port"
|
|
||||||
val streamProtocol = runCatching { StreamProtocol.valueOf(protocol.ifBlank { "PROGRESSIVE" }) }
|
|
||||||
.getOrDefault(StreamProtocol.PROGRESSIVE)
|
|
||||||
return when (streamProtocol) {
|
|
||||||
StreamProtocol.HLS, StreamProtocol.DASH -> "${baseUrl.trimEnd('/')}/manifest/$trackId"
|
|
||||||
StreamProtocol.PROGRESSIVE -> "${baseUrl.trimEnd('/')}/stream/$trackId"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private fun QueueEntry.matchesEntry(other: QueueEntry): Boolean {
|
private fun QueueEntry.matchesEntry(other: QueueEntry): Boolean {
|
||||||
return when {
|
return when {
|
||||||
this is QueueEntry.StreamingTrack && other is QueueEntry.StreamingTrack ->
|
this is QueueEntry.StreamingTrack && other is QueueEntry.StreamingTrack ->
|
||||||
|
|||||||
@ -22,6 +22,7 @@ import com.appstractive.dnssd.NetService
|
|||||||
import dev.krtirtho.spotube.core.discovery.DeviceDiscoveryService
|
import dev.krtirtho.spotube.core.discovery.DeviceDiscoveryService
|
||||||
import dev.krtirtho.spotube.core.server.LocalServer
|
import dev.krtirtho.spotube.core.server.LocalServer
|
||||||
import dev.krtirtho.spotube.modules.settings.SettingsRepository
|
import dev.krtirtho.spotube.modules.settings.SettingsRepository
|
||||||
|
import kotlinx.coroutines.CancellationException
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
import kotlinx.coroutines.IO
|
import kotlinx.coroutines.IO
|
||||||
@ -61,7 +62,12 @@ class RemoteControlService(
|
|||||||
val localDeviceId: StateFlow<String> = _localDeviceId.asStateFlow()
|
val localDeviceId: StateFlow<String> = _localDeviceId.asStateFlow()
|
||||||
|
|
||||||
private var advertisedService: NetService? = null
|
private var advertisedService: NetService? = null
|
||||||
|
|
||||||
|
/** A registration attempt that may have been left pending by the platform. */
|
||||||
|
private var pendingService: NetService? = null
|
||||||
|
|
||||||
private var registerJob: Job? = null
|
private var registerJob: Job? = null
|
||||||
|
private var cleanupJob: Job? = null
|
||||||
|
|
||||||
init {
|
init {
|
||||||
// Ensure a stable device id exists and is persisted up front, so discovery
|
// Ensure a stable device id exists and is persisted up front, so discovery
|
||||||
@ -112,6 +118,9 @@ class RemoteControlService(
|
|||||||
val port = localServer.port.value
|
val port = localServer.port.value
|
||||||
if (!settings.allowRemoteControl || port == null) return
|
if (!settings.allowRemoteControl || port == null) return
|
||||||
registerJob?.cancel()
|
registerJob?.cancel()
|
||||||
|
// Clean up any in-flight registration the cancelled job may have leaked.
|
||||||
|
scheduleCleanup(pendingService)
|
||||||
|
pendingService = null
|
||||||
registerJob = scope.launch {
|
registerJob = scope.launch {
|
||||||
registerLoop(settings.remoteControlDeviceName, port)
|
registerLoop(settings.remoteControlDeviceName, port)
|
||||||
}
|
}
|
||||||
@ -127,24 +136,53 @@ class RemoteControlService(
|
|||||||
attempt++
|
attempt++
|
||||||
// The user may have toggled the setting off during backoff.
|
// The user may have toggled the setting off during backoff.
|
||||||
if (!settingsRepository.userSettings.value.allowRemoteControl) return
|
if (!settingsRepository.userSettings.value.allowRemoteControl) return
|
||||||
|
val service = discoveryService.createService(
|
||||||
|
name = serviceName,
|
||||||
|
port = port,
|
||||||
|
deviceId = deviceId,
|
||||||
|
)
|
||||||
|
pendingService = service
|
||||||
try {
|
try {
|
||||||
advertisedService = discoveryService.advertise(
|
service.register(timeoutInMs = REGISTER_TIMEOUT_MS)
|
||||||
name = serviceName,
|
pendingService = null
|
||||||
port = port,
|
advertisedService = service
|
||||||
deviceId = deviceId,
|
|
||||||
registerTimeoutMs = REGISTER_TIMEOUT_MS,
|
|
||||||
)
|
|
||||||
log.i { "Advertising remote control service '$serviceName' on port $port (attempt $attempt)" }
|
log.i { "Advertising remote control service '$serviceName' on port $port (attempt $attempt)" }
|
||||||
|
} catch (e: CancellationException) {
|
||||||
|
throw e
|
||||||
} catch (e: Exception) {
|
} catch (e: Exception) {
|
||||||
|
pendingService = null
|
||||||
log.w(e) { "Failed to advertise remote control service (attempt $attempt); retrying in ${retryDelayMs(attempt)}ms" }
|
log.w(e) { "Failed to advertise remote control service (attempt $attempt); retrying in ${retryDelayMs(attempt)}ms" }
|
||||||
|
// The library leaks the platform registration on timeout. Once the
|
||||||
|
// platform eventually completes it (success), isRegistered flips
|
||||||
|
// and unregister() will actually remove it — keep trying until then.
|
||||||
|
scheduleCleanup(service)
|
||||||
delay(retryDelayMs(attempt))
|
delay(retryDelayMs(attempt))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Repeatedly tries to unregister a service whose registration attempt failed.
|
||||||
|
* The library's `unregister()` is a no-op while the platform hasn't completed
|
||||||
|
* the registration, so poll until it has (or give up after a while).
|
||||||
|
*/
|
||||||
|
private fun scheduleCleanup(service: NetService?) {
|
||||||
|
if (service == null) return
|
||||||
|
cleanupJob?.cancel()
|
||||||
|
cleanupJob = scope.launch {
|
||||||
|
repeat(REGISTER_CLEANUP_TRIES) {
|
||||||
|
delay(1_000)
|
||||||
|
runCatching { service.unregister() }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private suspend fun stopAdvertising() {
|
private suspend fun stopAdvertising() {
|
||||||
registerJob?.cancel()
|
registerJob?.cancel()
|
||||||
registerJob = null
|
registerJob = null
|
||||||
|
// Clean up any in-flight registration the cancelled job may have leaked.
|
||||||
|
scheduleCleanup(pendingService)
|
||||||
|
pendingService = null
|
||||||
if (advertisedService != null) {
|
if (advertisedService != null) {
|
||||||
runCatching { advertisedService?.unregister() }
|
runCatching { advertisedService?.unregister() }
|
||||||
advertisedService = null
|
advertisedService = null
|
||||||
@ -172,6 +210,12 @@ class RemoteControlService(
|
|||||||
}
|
}
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
private const val REGISTER_TIMEOUT_MS = 4_000L
|
// Generous enough that the library's timeout (which leaks the platform
|
||||||
|
// registration) rarely fires on a working system — registration callbacks
|
||||||
|
// normally arrive within a second.
|
||||||
|
private const val REGISTER_TIMEOUT_MS = 10_000L
|
||||||
|
|
||||||
|
// How long to keep polling unregister() on a failed service, in seconds.
|
||||||
|
private const val REGISTER_CLEANUP_TRIES = 15
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Loading…
Reference in New Issue
Block a user