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 fe9b97d2..35e9d432 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 @@ -34,9 +34,14 @@ class SubsonicSource( private val maxBitrate: Int = 0, private val streamFormat: String? = null, private val maxPages: Int = MAX_PAGES, + private val maxConcurrentFetches: Int = MAX_CONCURRENT_FETCHES, ) : MusicSource { private var emptyQuerySearchSupported: Boolean? = null + init { + require(maxConcurrentFetches > 0) { "maxConcurrentFetches must be positive" } + } + override suspend fun ping() { withContext(Dispatchers.IO) { client.check("ping") } } @@ -109,7 +114,7 @@ class SubsonicSource( override fun playlistSongs(playlistIds: List): Flow> = channelFlow { - val permits = Semaphore(MAX_CONCURRENT_FETCHES) + val permits = Semaphore(maxConcurrentFetches) playlistIds.forEach { playlistId -> launch { permits.withPermit { @@ -191,7 +196,7 @@ class SubsonicSource( private fun albumSongs(): Flow> = channelFlow { - val permits = Semaphore(MAX_CONCURRENT_FETCHES) + val permits = Semaphore(maxConcurrentFetches) var offset = 0 var pages = 0 while (true) { 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 874c3f3d..7e4e7905 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 @@ -1,5 +1,6 @@ package com.subtracks.data.source.subsonic +import com.subtracks.data.model.PlaylistSong import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.toList import kotlinx.coroutines.runBlocking @@ -315,6 +316,123 @@ class SubsonicSourceTest { assertTrue(maxInFlight.get() in 2..MAX_CONCURRENT_FETCHES) } + @Test + fun playlistSongsFetchesSequentiallyWhenTheBoundIsOne() = + runBlocking { + val inFlight = AtomicInteger() + val maxInFlight = AtomicInteger() + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + if (url.encodedPath != "/rest/getPlaylist.view") return MockResponse().setResponseCode(404) + val current = inFlight.incrementAndGet() + maxInFlight.updateAndGet { maxOf(it, current) } + Thread.sleep(50) + inFlight.decrementAndGet() + return MockResponse().setBody(playlistResponse(url.queryParameter("id")!!)) + } + } + + val entries = + SubsonicSource(1, client(), maxConcurrentFetches = 1) + .playlistSongs(listOf("p1", "p2", "p3", "p4")) + .toList() + .flatten() + + assertEquals(12, entries.size) + assertEquals(1, maxInFlight.get()) + } + + @Test + fun playlistSongsRespectsASmallerFetchBound() = + runBlocking { + val bound = 2 + val inFlight = AtomicInteger() + val maxInFlight = AtomicInteger() + val moreThanBoundStarted = CountDownLatch(bound + 1) + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + if (url.encodedPath != "/rest/getPlaylist.view") return MockResponse().setResponseCode(404) + val current = inFlight.incrementAndGet() + maxInFlight.updateAndGet { maxOf(it, current) } + moreThanBoundStarted.countDown() + moreThanBoundStarted.await(300, TimeUnit.MILLISECONDS) + inFlight.decrementAndGet() + return MockResponse().setBody(playlistResponse(url.queryParameter("id")!!)) + } + } + + val entries = + SubsonicSource(1, client(), maxConcurrentFetches = bound) + .playlistSongs(listOf("p1", "p2", "p3", "p4")) + .toList() + .flatten() + + assertEquals(12, entries.size) + assertEquals(bound, maxInFlight.get()) + } + + @Test + fun playlistSongsUseTheSameRowsAtAnyFetchBound() = + runBlocking { + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + if (url.encodedPath != "/rest/getPlaylist.view") return MockResponse().setResponseCode(404) + return MockResponse().setBody(playlistResponse(url.queryParameter("id")!!)) + } + } + val ids = listOf("p1", "p2", "p3", "p4") + + val sequential = SubsonicSource(1, client(), maxConcurrentFetches = 1).playlistSongs(ids).toList().flatten() + val concurrent = SubsonicSource(1, client(), maxConcurrentFetches = 4).playlistSongs(ids).toList().flatten() + + assertEquals(normalized(sequential), normalized(concurrent)) + } + + @Test + fun albumSongsFetchesSequentiallyWhenTheBoundIsOne() = + runBlocking { + val inFlight = AtomicInteger() + val maxInFlight = AtomicInteger() + server.dispatcher = + object : Dispatcher() { + override fun dispatch(request: RecordedRequest): MockResponse { + val url = request.requestUrl!! + return when { + url.encodedPath == "/rest/search3.view" -> { + MockResponse().setBody(failed(10, "Required parameter is missing")) + } + + url.encodedPath == "/rest/getAlbumList2.view" -> { + MockResponse().setBody(albumListOf(4)) + } + + url.encodedPath == "/rest/getAlbum.view" -> { + val current = inFlight.incrementAndGet() + maxInFlight.updateAndGet { maxOf(it, current) } + Thread.sleep(50) + inFlight.decrementAndGet() + MockResponse().setBody(albumWithSong(url.queryParameter("id")!!)) + } + + else -> { + MockResponse().setResponseCode(404) + } + } + } + } + + val songs = SubsonicSource(1, client(), maxConcurrentFetches = 1).songs().toList().flatten() + + assertEquals(4, songs.size) + assertEquals(1, maxInFlight.get()) + } + @Test fun playlistSongsPropagatesFailuresFromAnyPlaylist() = runBlocking { @@ -445,6 +563,26 @@ class SubsonicSourceTest { http = OkHttpClient(), ) + private fun playlistResponse(playlistId: String) = + "" + + (1..3).joinToString("") { "" } + + "" + + private fun normalized(entries: List): List = + entries.sortedWith(compareBy({ it.playlistId }, { it.position })) + + private fun albumListOf(count: Int) = + "" + + (0 until count).joinToString("") { + "" + } + + "" + + private fun albumWithSong(albumId: String) = + "" + + "" + + "" + private fun failed( code: Int, message: String,