make the sync fetch concurrency bound configurable

This commit is contained in:
austinried
2026-10-04 00:30:01 +00:00
parent 8ba9e92fa6
commit dd092ed7ad
2 changed files with 145 additions and 2 deletions
@@ -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<String>): Flow<List<PlaylistSong>> =
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<List<Song>> =
channelFlow {
val permits = Semaphore(MAX_CONCURRENT_FETCHES)
val permits = Semaphore(maxConcurrentFetches)
var offset = 0
var pages = 0
while (true) {
@@ -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) =
"<subsonic-response status=\"ok\"><playlist id=\"$playlistId\">" +
(1..3).joinToString("") { "<entry id=\"$playlistId-s$it\" title=\"Title $it\"/>" } +
"</playlist></subsonic-response>"
private fun normalized(entries: List<PlaylistSong>): List<PlaylistSong> =
entries.sortedWith(compareBy({ it.playlistId }, { it.position }))
private fun albumListOf(count: Int) =
"<subsonic-response status=\"ok\"><albumList2>" +
(0 until count).joinToString("") {
"<album id=\"al$it\" name=\"Album $it\" artist=\"Artist\" artistId=\"ar1\" songCount=\"1\"/>"
} +
"</albumList2></subsonic-response>"
private fun albumWithSong(albumId: String) =
"<subsonic-response status=\"ok\"><album id=\"$albumId\">" +
"<song id=\"$albumId-s1\" title=\"One\"/>" +
"</album></subsonic-response>"
private fun failed(
code: Int,
message: String,