diff --git a/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicClient.kt b/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicClient.kt index 289ec10d..ed034c77 100644 --- a/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicClient.kt +++ b/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicClient.kt @@ -6,7 +6,7 @@ import okhttp3.Request import java.io.InputStream import java.security.MessageDigest -class SubsonicException( +open class SubsonicException( val code: Int, message: String, ) : Exception(message) diff --git a/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt b/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt index 178d2628..22888fed 100644 --- a/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt +++ b/app/src/main/kotlin/com/subtracks/data/source/subsonic/SubsonicSource.kt @@ -16,6 +16,7 @@ import kotlinx.coroutines.flow.channelFlow import kotlinx.coroutines.flow.emitAll import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOn +import kotlinx.coroutines.flow.map import kotlinx.coroutines.withContext import org.xml.sax.Attributes @@ -38,8 +39,7 @@ class SubsonicSource( method = "getArtists", tag = "artist", create = { SubsonicXml.artist(id, it) }, - accept = { it.id.isNotEmpty() }, - ) + ).map { batch -> batch.filter { it.id.isNotEmpty() } } override fun albums(): Flow> = flow { @@ -47,26 +47,31 @@ class SubsonicSource( var offset = 0 var pages = 0 while (true) { - var count = 0 + var raw = 0 entityBatches( method = "getAlbumList2", params = page("newest", offset), tag = "album", create = { SubsonicXml.albumDraft(id, it) }, onChild = { draft, name, attrs -> if (name == "discTitles") draft.addDiscTitle(attrs) }, - accept = { it.id.isNotEmpty() }, ).collect { batch -> - count += batch.size - emit( - batch.map { - it.toAlbum().copy( - frequentRank = frequent[it.id], - recentRank = recent[it.id], - ) - }, - ) + raw += batch.size + val accepted = batch.filter { it.id.isNotEmpty() } + if (accepted.isNotEmpty()) { + emit( + accepted.map { + it.toAlbum().copy( + frequentRank = frequent[it.id], + recentRank = recent[it.id], + ) + }, + ) + } + } + if (raw < PAGE_SIZE) { + if (raw > 0 && pages >= maxPages) throw pageCapExceeded() + break } - if (count < PAGE_SIZE) break pages++ if (pages > maxPages) throw pageCapExceeded() offset += PAGE_SIZE @@ -87,8 +92,7 @@ class SubsonicSource( method = "getPlaylists", tag = "playlist", create = { SubsonicXml.playlist(id, it) }, - accept = { it.id.isNotEmpty() }, - ) + ).map { batch -> batch.filter { it.id.isNotEmpty() } } override fun playlistSongs(playlistIds: List): Flow> = flow { @@ -102,8 +106,10 @@ class SubsonicSource( val entryId = attrs.attr("id") SubsonicXml.playlistSong(id, playlistId, if (entryId.isEmpty()) -1L else position++, attrs) }, - accept = { it.songId.isNotEmpty() }, - ).collect { emit(it) } + ).collect { batch -> + val accepted = batch.filter { it.songId.isNotEmpty() } + if (accepted.isNotEmpty()) emit(accepted) + } } }.flowOn(Dispatchers.IO).buffer(1) @@ -139,7 +145,7 @@ class SubsonicSource( var offset = 0 var pages = 0 while (true) { - var count = 0 + var raw = 0 entityBatches( method = "search3", params = @@ -152,12 +158,15 @@ class SubsonicSource( ), tag = "song", create = { SubsonicXml.song(id, it) }, - accept = { it.id.isNotEmpty() }, ).collect { batch -> - count += batch.size - emit(batch) + raw += batch.size + val accepted = batch.filter { it.id.isNotEmpty() } + if (accepted.isNotEmpty()) emit(accepted) + } + if (raw < PAGE_SIZE) { + if (raw > 0 && pages >= maxPages) throw pageCapExceeded() + break } - if (count < PAGE_SIZE) break pages++ if (pages > maxPages) throw pageCapExceeded() offset += PAGE_SIZE @@ -169,24 +178,32 @@ class SubsonicSource( var offset = 0 var pages = 0 while (true) { + var raw = 0 val albums = ArrayList(PAGE_SIZE) entityBatches( method = "getAlbumList2", params = page("alphabeticalByName", offset), tag = "album", create = { it.attr("id") }, - accept = { it.isNotEmpty() }, - ).collect { albums += it } + ).collect { batch -> + raw += batch.size + batch.forEach { if (it.isNotEmpty()) albums += it } + } for (albumId in albums) { entityBatches( method = "getAlbum", params = mapOf("id" to albumId), tag = "song", create = { SubsonicXml.song(id, it) }, - accept = { it.id.isNotEmpty() }, - ).collect { emit(it) } + ).collect { batch -> + val accepted = batch.filter { it.id.isNotEmpty() } + if (accepted.isNotEmpty()) emit(accepted) + } + } + if (raw < PAGE_SIZE) { + if (raw > 0 && pages >= maxPages) throw pageCapExceeded() + break } - if (albums.size < PAGE_SIZE) break pages++ if (pages > maxPages) throw pageCapExceeded() offset += PAGE_SIZE @@ -201,26 +218,33 @@ class SubsonicSource( var offset = 0 var pages = 0 while (true) { + var raw = 0 var index = 0 entityBatches( method = "getAlbumList2", params = page(type, offset), tag = "album", create = { it.attr("id") }, - accept = { it.isNotEmpty() }, ).collect { batch -> + raw += batch.size batch.forEach { albumId -> - ranks[albumId] = (offset + index).toLong() - index++ + if (albumId.isNotEmpty()) { + ranks[albumId] = (offset + index).toLong() + index++ + } } } - if (index < PAGE_SIZE) break + if (raw < PAGE_SIZE) { + if (raw > 0 && pages >= maxPages) throw pageCapExceeded() + break + } pages++ if (pages > maxPages) throw pageCapExceeded() offset += PAGE_SIZE } ranks - } catch (_: SubsonicException) { + } catch (failure: SubsonicException) { + if (failure is PageCapExceeded) throw failure emptyMap() } @@ -239,7 +263,7 @@ class SubsonicSource( return supported } - private fun pageCapExceeded() = SubsonicException(-1, "Library exceeds the ${maxPages * PAGE_SIZE}-row page cap") + private fun pageCapExceeded() = PageCapExceeded("Library exceeds the ${maxPages * PAGE_SIZE}-row page cap") private fun entityBatches( method: String, @@ -247,7 +271,6 @@ class SubsonicSource( tag: String, create: (Attributes) -> T, onChild: (T, String, Attributes) -> Unit = { _, _, _ -> }, - accept: (T) -> Boolean = { true }, ): Flow> = channelFlow { val buffer = ArrayList(PAGE_SIZE) @@ -267,10 +290,8 @@ class SubsonicSource( create = create, onChild = onChild, onEnd = { entity -> - if (accept(entity)) { - buffer += entity - if (buffer.size >= PAGE_SIZE) flush() - } + buffer += entity + if (buffer.size >= PAGE_SIZE) flush() }, ) } @@ -293,3 +314,7 @@ class SubsonicSource( const val THUMBNAIL_SIZE = 256 } } + +private class PageCapExceeded( + message: String, +) : SubsonicException(-1, message) diff --git a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt index 8e90af65..07f042a2 100644 --- a/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt +++ b/app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt @@ -8,10 +8,10 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex sealed interface SyncStatus { data object Idle : SyncStatus @@ -31,42 +31,34 @@ class SyncManager( private val queueRepository: QueueRepository, ) { private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - private val syncLock = Mutex() + private val requests = Channel(Channel.CONFLATED) private val _status = MutableStateFlow(SyncStatus.Idle) val status: StateFlow = _status - fun requestSync() { - if (!syncLock.tryLock()) return + init { scope.launch { - try { - runSync() - } finally { - syncLock.unlock() - } + for (ignored in requests) runSync() } } + fun requestSync() { + requests.trySend(Unit) + } + private suspend fun runSync() { _status.value = SyncStatus.Running - val result = + _status.value = try { val source = sourceRepository.activeMusicSource() ?: error("No server configured") SyncService(db, source).sync() - Result.success(Unit) + SyncStatus.Success } catch (cancellation: CancellationException) { throw cancellation } catch (failure: Exception) { - Result.failure(failure) + Log.w(TAG, "Sync failed", failure) + SyncStatus.Failed(failure.message ?: "Sync failed") } queueRepository.invalidateLibraryCache() - _status.value = - result.fold( - onSuccess = { SyncStatus.Success }, - onFailure = { error -> - Log.w(TAG, "Sync failed", error) - SyncStatus.Failed(error.message ?: "Sync failed") - }, - ) } private companion object { diff --git a/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt b/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt index 04cf5749..87e780ea 100644 --- a/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/source/subsonic/SubsonicSourceTest.kt @@ -224,6 +224,52 @@ class SubsonicSourceTest { assertEquals(1000, albums.size) } + @Test + fun albumsAbortWhenThePageJustOverTheCapIsNotEmpty() = + runBlocking { + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + if (url.encodedPath != "/rest/getAlbumList2.view") return MockResponse().setResponseCode(404) + if (url.queryParameter("type") != "newest") return MockResponse().setBody(emptyAlbumList()) + return when (url.queryParameter("offset")) { + "0", "500" -> MockResponse().setBody(albumPage(url.queryParameter("offset")!!.toInt(), 500)) + else -> MockResponse().setBody(albumPage(1000, 1)) + } + } + } + + val failure = + runCatching { + SubsonicSource(1, client(), maxPages = 2).albums().toList() + } + + assertEquals(true, failure.exceptionOrNull() is SubsonicException) + } + + @Test + fun emptyIdRowsDoNotEndPaginationEarly() = + runBlocking { + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + if (url.encodedPath != "/rest/getAlbumList2.view") return MockResponse().setResponseCode(404) + if (url.queryParameter("type") != "newest") return MockResponse().setBody(emptyAlbumList()) + return when (url.queryParameter("offset")) { + "0" -> MockResponse().setBody(albumPageWithEmptyId(0, 500)) + else -> MockResponse().setBody(albumPage(500, 3)) + } + } + } + + val albums = SubsonicSource(1, client()).albums().toList().flatten() + + assertEquals(502, albums.size) + assertEquals("al-502", albums.last().id) + } + @Test fun playlistEntriesWithoutIdsAreSkipped() = runBlocking { @@ -293,6 +339,16 @@ class SubsonicSourceTest { } + "" + private fun albumPageWithEmptyId( + offset: Int, + count: Int, + ) = "" + + "" + + (offset until offset + count - 1).joinToString("") { + "" + } + + "" + private fun songPage( prefix: String, offset: Int, diff --git a/docs/architecture.md b/docs/architecture.md index f7c39c7b..4cb34959 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -106,7 +106,7 @@ Images: grids and lists request a 256 px `getCoverArt` thumbnail; the 120 dp det ## Sync -`SyncService.sync()` is serialized by a mutex in `SourceRepository`, so overlapping sync requests cannot interleave. It walks five entities in sequence — artists, albums, songs, playlists and playlist songs — and each one is a cold `Flow` from `MusicSource` that the service collects: it remembers the IDs already in the local mirror, upserts every batch as it arrives (one `immediateTransaction` per ~500-row batch), crosses those IDs off, and after that entity's stream ends deletes whatever is left, i.e. the rows the server no longer lists. The parser reads the HTTP response stream directly, so neither the raw XML nor the whole library ever sits in memory at once. +`SyncService.sync()` is serialized by `SyncManager`, which coalesces duplicate requests. It walks five entities in sequence — artists, albums, songs, playlists and playlist songs — and each one is a cold `Flow` from `MusicSource` that the service collects: it remembers the IDs already in the local mirror, upserts every batch as it arrives (one `immediateTransaction` per ~500-row batch), crosses those IDs off, and after that entity's stream ends deletes whatever is left, i.e. the rows the server no longer lists. The parser reads the HTTP response stream directly, so the fetched library is streamed page by page rather than materialised wholesale, though the prune still holds the existing ID set to diff against. Songs have a primary and a fallback fetch. The primary probes `search3` with an empty query once (memoized per source); if the server supports it, the library is paged through `search3` 500 songs at a time. Otherwise the fallback pages `getAlbumList2` and then issues one `getAlbum` per album — it works everywhere but costs a request per album instead of per 500 songs. Every paged loop stops on a short page and is capped at 1000 pages. Playlists are fetched whole, and their songs are stored by position so a playlist that got shorter (or was emptied) is trimmed at the end.