diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/IptvRepository.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/IptvRepository.kt index e70fbc3f5..282dcf517 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/IptvRepository.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/IptvRepository.kt @@ -360,6 +360,47 @@ class IptvRepository @Inject constructor( @Volatile private var cachedStalkerApis: Map = emptyMap() + /** + * Scope for the shared Stalker channel-list download. Deliberately not tied + * to a caller: when the entry point that started the download gives up (its + * own timeout, a screen the user left), the download still finishes and the + * next entry point reuses it instead of starting another one. + */ + private val stalkerChannelListScope = kotlinx.coroutines.CoroutineScope( + kotlinx.coroutines.SupervisorJob() + Dispatchers.IO + ) + + /** + * What one portal answered: the session it opened and the channels it + * returned. A portal that did not answer carries a null [api] and no + * channels — that is what makes a failure visible per portal instead of + * disappearing into a merged list. + */ + internal data class StalkerPortalChannels( + val portalId: String, + val api: com.arflix.tv.data.api.StalkerApi?, + val channels: List + ) + + /** + * One shared download **per portal**. + * + * Keyed per portal rather than per portal set: with several portals + * configured, a portal that is temporarily down must not hide behind one + * that answered. Its result is never remembered, so the next ordinary load + * asks it again, while the portals that did answer keep their lists. + * + * `internal` so a test can check what [invalidateCache] and + * [ensureCacheOwnership] do to it — the bug this loader exists to prevent + * came back through its caller, not through the loader itself (same + * convention as [activePlaylists]). + */ + internal val stalkerChannelListLoader = + StalkerChannelListLoader( + scope = stalkerChannelListScope, + isReusable = { it.api != null && it.channels.isNotEmpty() } + ) + private data class StalkerEpgPortalCacheKey( val portalId: String, val apiIdentity: String @@ -594,6 +635,93 @@ class IptvRepository @Inject constructor( private fun activeStalkerPortals(config: IptvConfig): List = config.stalkerPortals.filter { it.enabled && it.portalUrl.isNotBlank() } + /** + * Identifies one configured portal for [stalkerChannelListLoader]. Hashed so + * the portal URL and MAC address never travel further than this function. + */ + private fun stalkerPortalKey(portal: StalkerPortalEntry): String { + val raw = listOf(portal.id.trim(), portal.portalUrl.trim(), portal.macAddress.trim()) + .joinToString("|") + return MessageDigest.getInstance("SHA-256") + .digest(raw.toByteArray(StandardCharsets.UTF_8)) + .joinToString("") { "%02x".format(it) } + } + + /** The loader keys of the portals [config] currently has enabled. */ + private fun activeStalkerPortalKeys(config: IptvConfig): Set = + activeStalkerPortals(config).map { stalkerPortalKey(it) }.toSet() + + /** + * Downloads every enabled portal's channel list, each portal at most once + * per freshness window, and merges the answers. + * + * This is the only place that opens portal sessions for the channel list — + * the startup prefetch, the live TV snapshot load and the guide backfill + * load all come through here, so a start costs one download instead of one + * per entry point (measured before: 3 x 29 MB in 17 seconds). Sharing and + * the freshness window live in [StalkerChannelListLoader]. + * + * Each portal is loaded under its own key, so a portal that fails is + * retried by the next ordinary load while the portals that answered keep + * their lists. Merging is the only thing that happens here, and it happens + * after the decision what may be remembered — that decision is per portal. + * + * Channel ids are prefixed with `stalker::` so playback + * can route back to the portal that owns them. + * + * `internal` so a test can drive the real merge and the real loader keys + * with [fetchPortal] standing in for the network; production never passes + * it. + * + * @param freshSinceMs the moment the caller decided it needed fresh data; + * `0` accepts any list inside the freshness window. See + * [StalkerChannelListLoader.load]. + */ + internal suspend fun loadStalkerChannels( + portals: List, + freshSinceMs: Long = 0L, + fetchPortal: suspend (StalkerPortalEntry) -> StalkerPortalChannels = { fetchStalkerChannels(it) } + ): Pair, List> = coroutineScope { + if (portals.isEmpty()) { + return@coroutineScope emptyMap() to emptyList() + } + val answers = portals + .map { portal -> + async { + stalkerChannelListLoader.load(stalkerPortalKey(portal), freshSinceMs) { + fetchPortal(portal) + } + } + } + .awaitAll() + val apis = LinkedHashMap() + val channels = ArrayList() + for (answer in answers) { + channels.addAll(answer.channels) + answer.api?.let { apis[answer.portalId] = it } + } + apis.toMap() to channels.toList() + } + + /** + * Opens one portal's session and downloads its channels. A portal that does + * not answer returns no session and no channels, which is what keeps its + * failure out of [stalkerChannelListLoader]'s memory. + */ + private suspend fun fetchStalkerChannels(portal: StalkerPortalEntry): StalkerPortalChannels = + runCatching { + val stalker = com.arflix.tv.data.api.StalkerApi(portal.portalUrl, portal.macAddress) + if (!stalker.handshake()) { + return@runCatching StalkerPortalChannels(portal.id, null, emptyList()) + } + stalker.getProfile() + StalkerPortalChannels( + portalId = portal.id, + api = stalker, + channels = stalker.getChannels().map { it.copy(id = "stalker:${portal.id}:${it.id}") } + ) + }.getOrElse { StalkerPortalChannels(portal.id, null, emptyList()) } + @Volatile private var xtreamSeriesLoadedAtMs: Long = 0L @Volatile @@ -2081,6 +2209,12 @@ class IptvRepository @Inject constructor( onProgress: (IptvLoadProgress) -> Unit = {}, onChannelsReady: suspend (List) -> Unit = {} ): IptvSnapshot { + // Taken before the lock on purpose: it is the moment this caller asked + // for data, not the moment it got its turn. A forced reload uses it to + // tell "the list is from before I asked" from "someone downloaded it + // while I was waiting" — the second entry point reacting to a single + // configuration change is the latter, and must not download again. + val requestedAtMs = System.currentTimeMillis() return withContext(Dispatchers.IO) { loadMutex.withLock { cleanupStaleEpgTempFiles() @@ -2107,32 +2241,20 @@ class IptvRepository @Inject constructor( // Load every enabled Stalker portal in parallel with M3U/Xtream // playlists when both are configured (hybrid mode). Stalker-only // (no playlists) keeps the legacy early-return behavior. + // A forced reload asks for data that is newer than the request, so + // a list remembered from before it is skipped — but one downloaded + // while this caller waited for the lock counts, and a download that + // is already running is shared. Skipping every remembered list + // instead made two entry points reacting to the same configuration + // change pull the full channel list twice (measured: 2 x 27.67 MB + // on every playlist toggle). val stalkerChannelsDeferred = if (stalkerPortals.isNotEmpty()) { async { onProgress(IptvLoadProgress(context.getString(R.string.iptv_connecting_stalker), 10)) - stalkerPortals.map { portal -> - async { - runCatching { - val stalker = com.arflix.tv.data.api.StalkerApi(portal.portalUrl, portal.macAddress) - if (!stalker.handshake()) { - return@runCatching Triple>(portal.id, null, emptyList()) - } - stalker.getProfile() - val channels = stalker.getChannels() - // Prefix with stalker:: so the - // portal can be identified for playback routing. - Triple>(portal.id, stalker, channels.map { it.copy(id = "stalker:${portal.id}:${it.id}") }) - }.getOrElse { Triple>(portal.id, null, emptyList()) } - } - }.awaitAll().let { results -> - val apis = HashMap() - val channels = ArrayList() - for ((portalId, api, chs) in results) { - channels.addAll(chs) - api?.let { apis[portalId] = it } - } - apis to channels - } + loadStalkerChannels( + stalkerPortals, + freshSinceMs = if (forcePlaylistReload) requestedAtMs else 0L + ) } } else { null @@ -3581,44 +3703,14 @@ class IptvRepository @Inject constructor( // Stalker-only mode: no playlists configured. if (activeLists.isEmpty() && stalkerPortals.isNotEmpty()) { - val apis = HashMap() - val channels = ArrayList() - for (portal in stalkerPortals) { - runCatching { - val stalker = com.arflix.tv.data.api.StalkerApi(portal.portalUrl, portal.macAddress) - if (!stalker.handshake()) return@runCatching - stalker.getProfile() - stalker.getChannels().map { it.copy(id = "stalker:${portal.id}:${it.id}") } - .also { channels.addAll(it) } - apis[portal.id] = stalker - } - } + val (apis, channels) = loadStalkerChannels(stalkerPortals) return if (channels.isNotEmpty()) channels to apis else null } // Load playlists and Stalker in parallel when both are configured. val (playlistChannels, stalkerApis, stalkerChannels) = coroutineScope { val stalkerDeferred = if (stalkerPortals.isNotEmpty()) { - async { - stalkerPortals.map { portal -> - async { - runCatching { - val stalker = com.arflix.tv.data.api.StalkerApi(portal.portalUrl, portal.macAddress) - if (!stalker.handshake()) return@runCatching Triple>(portal.id, null, emptyList()) - stalker.getProfile() - Triple>(portal.id, stalker, stalker.getChannels().map { it.copy(id = "stalker:${portal.id}:${it.id}") }) - }.getOrElse { Triple>(portal.id, null, emptyList()) } - } - }.awaitAll().let { results -> - val apis = HashMap() - val channels = ArrayList() - for ((portalId, api, chs) in results) { - channels.addAll(chs) - api?.let { apis[portalId] = it } - } - apis to channels - } - } + async { loadStalkerChannels(stalkerPortals) } } else { null } @@ -3693,6 +3785,15 @@ class IptvRepository @Inject constructor( } fun invalidateCache() { + // The shared Stalker downloads are deliberately NOT dropped here. Each + // is keyed by one portal (id + URL + MAC), and portals that disappear + // are released in ensureCacheOwnership, which every entry point passes + // through. Everything else this function is called for — a playlist + // toggled, an EPG URL edited, a profile switched to one with the same + // portals — leaves the portals alone, so their channel lists stay + // valid. Dropping them anyway tore up a download that another entry + // point was already running and cost a second full 27.67 MB list on + // every playlist toggle (measured). cachedChannels = emptyList() cachedChannelsLookupSource = null cachedChannelsById = emptyMap() @@ -3765,7 +3866,41 @@ class IptvRepository @Inject constructor( runCatching { channelCacheFile().delete() } } - private fun ensureCacheOwnership(profileId: String, config: IptvConfig) { + /** + * Releases what is held for Stalker portals that are no longer configured: + * their channel list and, with it, the portal session that came with it. + * + * [loadStalkerChannels] cannot do this. When the last portal is removed + * every caller returns before it is reached, so the removed portal's list + * and session stayed in memory until the app was closed. + * + * Only portals that actually disappeared are dropped. A playlist toggled, + * an EPG URL edited or a profile switched to one with the same portals + * leaves every key in place, so unrelated changes keep sharing the download + * they already have (measured: dropping it anyway cost a second full + * 27.67 MB list on every playlist toggle). + */ + private fun releaseStalkerStateForRemovedPortals(config: IptvConfig) { + val liveKeys = activeStalkerPortalKeys(config) + stalkerChannelListLoader.retainOnly(liveKeys) + val livePortalIds = activeStalkerPortals(config).map { it.id }.toSet() + if (cachedStalkerApis.keys.any { it !in livePortalIds }) { + cachedStalkerApis = cachedStalkerApis.filterKeys { it in livePortalIds } + } + } + + /** + * Runs before any entry point can decide it has nothing to do — the live TV + * snapshot load, the cache-only warmup and the cached-snapshot read all pass + * through here first, whether or not a source is configured. That makes it + * the one place where "the last portal is gone" is actually observed, so it + * is where the Stalker state of removed portals is released. + * + * `internal` so a test can drive it with a plain [IptvConfig], same + * convention as [activePlaylists]. + */ + internal fun ensureCacheOwnership(profileId: String, config: IptvConfig) { + releaseStalkerStateForRemovedPortals(config) val sig = buildSourceSignature(config) val ownerChanged = cacheOwnerProfileId != null && cacheOwnerProfileId != profileId val configChanged = cacheOwnerConfigSig != null && cacheOwnerConfigSig != sig diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/StalkerChannelListLoader.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/StalkerChannelListLoader.kt new file mode 100644 index 000000000..9b275ad19 --- /dev/null +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/StalkerChannelListLoader.kt @@ -0,0 +1,145 @@ +package com.arflix.tv.data.repository + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.async + +/** + * How long a downloaded Stalker channel list stays reusable inside the running + * app. Long enough to cover the startup burst and normal navigation between + * screens, short enough that the `play_token` carried in every channel URL of + * that list stays young. + */ +internal const val STALKER_CHANNEL_LIST_FRESHNESS_MS = 5 * 60_000L + +/** + * Lets every entry point that needs the Stalker channel list share a single + * download instead of starting its own. + * + * Three entry points ask for the list on a normal app start — the startup + * prefetch, the live TV snapshot load and the guide backfill load. Each used to + * open its own portal session and pull the full list; on a portal with ~21k + * channels that is 29 MB per entry point. + * + * Two mechanisms, both memory only: + * - concurrent callers await the download that is already running; + * - a caller arriving within [freshnessWindowMs] of a usable download reuses + * its result instead of starting another one. + * + * **One entry per key, and a key is one portal.** Every portal is downloaded, + * remembered and retried on its own, so the outcome of one portal can never + * decide anything about another. A portal that failed leaves no entry behind + * and is asked again by the very next load, while the portals that did answer + * keep their lists — which is the whole point of this class for anybody running + * more than one portal. + * + * A caller that wants genuinely fresh data passes `freshSinceMs` — the moment + * it decided it needed fresh data. A remembered list downloaded after that + * moment already answers the request; an older one does not and is skipped. + * That timestamp is what keeps one configuration change from costing two + * downloads: the second entry point reacting to it asked before the first + * one's download finished, so that download is fresh enough for it too. + * A download that is already running is joined either way — a running download + * is never older than the remembered result (a new one only starts when none + * is in flight), so joining it is what "fresh" means here. + * + * Nothing discards a running download except [retainOnly], which drops the + * portals that are no longer configured. A configuration change that leaves a + * portal alone has nothing to do with that portal's channel list. + * + * The download runs in [scope] rather than in the calling coroutine, so a + * caller that gives up (its own timeout, a screen the user left) does not + * cancel the download the other callers are waiting for. + * + * Deliberately **not** persisted across app runs: the playable URLs in the + * channel list carry a `play_token`, so a list restored from disk would hand + * out expired tokens. A fresh app start always downloads once. + * + * @param isReusable decides whether a result may be remembered. A portal whose + * handshake failed yields an empty list, and remembering that would keep the + * portal dark for the whole window. + */ +internal class StalkerChannelListLoader( + private val freshnessWindowMs: Long = STALKER_CHANNEL_LIST_FRESHNESS_MS, + private val scope: CoroutineScope, + private val nowMs: () -> Long = System::currentTimeMillis, + private val isReusable: (T) -> Boolean = { true } +) { + + private val lock = Any() + + /** One entry per key; see [retainOnly] for the only thing that removes one. */ + private val entries = HashMap>() + + private class Entry { + var inFlight: Deferred? = null + var remembered: T? = null + var rememberedAtMs: Long = 0L + } + + /** + * Returns the channel list for [key], running [fetch] at most once per + * freshness window. [key] identifies a single configured portal (id + URL + + * MAC), so portals never share an entry and never mask each other. + * + * @param freshSinceMs the moment the caller decided it needed fresh data. + * A remembered list downloaded before that moment is skipped; one + * downloaded after it already answers the request. Pass `0` to accept any + * list inside the freshness window. A download that is already running is + * shared in either case: asking for fresh data is about not being served + * an old list, not about opening a second portal session next to the + * first. + */ + suspend fun load(key: String, freshSinceMs: Long = 0L, fetch: suspend () -> T): T { + val pending = synchronized(lock) { + val entry = entries.getOrPut(key) { Entry() } + val previous = entry.remembered + if (previous != null) { + val insideWindow = nowMs() - entry.rememberedAtMs < freshnessWindowMs + if (insideWindow && entry.rememberedAtMs >= freshSinceMs) return previous + entry.remembered = null + entry.rememberedAtMs = 0L + } + entry.inFlight?.takeIf { !it.isCompleted } ?: startLocked(key, entry, fetch) + } + return pending.await() + } + + /** + * Drops everything held for portals outside [keys] — their channel list and, + * with it, the portal session that came with it. + * + * This is the only thing that discards an entry, and it exists for the case + * that never calls [load]: a portal was disabled or removed, so nothing asks + * for its channel list any more and it would otherwise be held until the app + * dies. Passing an empty set releases everything, which is what removing the + * *last* portal means. + * + * Portals that are still configured keep their entries, so an unrelated + * playlist change costs nothing. + */ + fun retainOnly(keys: Set) { + synchronized(lock) { + entries.keys.retainAll(keys) + } + } + + private fun startLocked(key: String, entry: Entry, fetch: suspend () -> T): Deferred { + val deferred = scope.async { + val value = fetch() + synchronized(lock) { + // Only store while this download is still the current one for + // its key. A download detached by [retainOnly] finds a removed + // (or freshly recreated) entry and drops its result instead of + // overwriting the one that replaced it. + if (entries[key] === entry && isReusable(value)) { + entry.remembered = value + entry.rememberedAtMs = nowMs() + } + } + value + } + entry.inFlight = deferred + return deferred + } +} diff --git a/app/src/test/kotlin/com/arflix/tv/data/repository/IptvRepositoryStalkerListInvalidationTest.kt b/app/src/test/kotlin/com/arflix/tv/data/repository/IptvRepositoryStalkerListInvalidationTest.kt new file mode 100644 index 000000000..5d43180dd --- /dev/null +++ b/app/src/test/kotlin/com/arflix/tv/data/repository/IptvRepositoryStalkerListInvalidationTest.kt @@ -0,0 +1,95 @@ +package com.arflix.tv.data.repository + +import com.arflix.tv.data.api.StalkerApi +import com.arflix.tv.data.model.IptvChannel +import com.google.common.truth.Truth.assertThat +import kotlinx.coroutines.runBlocking +import org.junit.Test + +/** + * The caller side of [StalkerChannelListLoader]. + * + * Measured on device on 09.09.2026: the loader itself was correct and its own + * tests were green, yet toggling a playlist still pulled the full 27.67 MB + * channel list twice. The second download came from one level up — + * [IptvRepository.invalidateCache] threw the shared download away, and every + * source change runs through it. + * + * So this test checks the caller, not the loader: a change that leaves the + * configured portals alone must not cost a second download. What discards a + * list is a portal disappearing from the configuration, and only that — see + * [IptvRepositoryStalkerPortalStateTest] for that side and + * [StalkerChannelListLoaderTest] for the loader itself. + */ +class IptvRepositoryStalkerListInvalidationTest { + + private fun newRepository(): IptvRepository { + val context = io.mockk.mockk(relaxed = true) + val okHttpClient = io.mockk.mockk(relaxed = true) + val profileManager = io.mockk.mockk(relaxed = true) + val invalidationBus = io.mockk.mockk(relaxed = true) + return IptvRepository(context, okHttpClient, profileManager, invalidationBus) + } + + private fun downloaded() = IptvRepository.StalkerPortalChannels( + portalId = "portal1", + api = io.mockk.mockk(relaxed = true), + channels = listOf( + IptvChannel( + id = "stalker:portal1:1", + name = "Channel 1", + streamUrl = "http://portal.invalid/play/live.php", + group = "News" + ) + ) + ) + + @Test + fun `invalidateCache keeps the shared Stalker channel list`() = runBlocking { + val repository = newRepository() + var downloads = 0 + + repository.stalkerChannelListLoader.load("portal-key") { downloads++; downloaded() } + // A playlist toggled, an EPG URL edited, a profile switched: every one + // of these lands here, and none of them touches the Stalker portals. + repository.invalidateCache() + repository.stalkerChannelListLoader.load("portal-key") { downloads++; downloaded() } + + assertThat(downloads).isEqualTo(1) + } + + @Test + fun `purgeAllIptvSourceCaches keeps the shared Stalker channel list`() = runBlocking { + // The explicit "Refresh IPTV" runs this before loading. Real fresh + // data comes from the load that follows it (it passes the moment the + // user asked), not from tearing up a download another entry point may + // be running right now. + val repository = newRepository() + var downloads = 0 + + repository.stalkerChannelListLoader.load("portal-key") { downloads++; downloaded() } + repository.purgeAllIptvSourceCaches(preserveLiveSnapshot = true) + repository.stalkerChannelListLoader.load("portal-key") { downloads++; downloaded() } + + assertThat(downloads).isEqualTo(1) + } + + @Test + fun `a forced load after invalidateCache still reaches the portal`() = runBlocking { + // The other half of the promise: keeping the list must not make + // "Refresh IPTV" serve stale channels. A caller that asks for data + // newer than its own request still downloads. + val repository = newRepository() + var downloads = 0 + + repository.stalkerChannelListLoader.load("portal-key") { downloads++; downloaded() } + repository.invalidateCache() + val askedAtMs = System.currentTimeMillis() + 1 + repository.stalkerChannelListLoader.load("portal-key", freshSinceMs = askedAtMs) { + downloads++ + downloaded() + } + + assertThat(downloads).isEqualTo(2) + } +} diff --git a/app/src/test/kotlin/com/arflix/tv/data/repository/IptvRepositoryStalkerPortalStateTest.kt b/app/src/test/kotlin/com/arflix/tv/data/repository/IptvRepositoryStalkerPortalStateTest.kt new file mode 100644 index 000000000..da46ec823 --- /dev/null +++ b/app/src/test/kotlin/com/arflix/tv/data/repository/IptvRepositoryStalkerPortalStateTest.kt @@ -0,0 +1,158 @@ +package com.arflix.tv.data.repository + +import com.arflix.tv.data.api.StalkerApi +import com.arflix.tv.data.model.IptvChannel +import com.google.common.truth.Truth.assertThat +import java.util.Collections +import kotlinx.coroutines.runBlocking +import org.junit.Test + +/** + * The two cases Prodigy asked for in his review of #677, both driven through + * the repository rather than through [StalkerChannelListLoader] alone — the + * loader's own 13 tests were green while both of these were broken one level up. + * + * 1. A partial failure must not be remembered as a success. With portals A and + * B configured and B down, the merged channel list is not empty (A filled + * it), so a freshness check that looks at the merged result calls the load + * reusable and stops asking B for the whole window — even after B recovers. + * 2. The cleanup for "the last portal was removed" must sit on a path that + * actually runs. [IptvRepository.loadStalkerChannels] is not one: every + * caller returns before reaching it once no portal is enabled, so the + * removed portal's channel list and session stayed in memory until the app + * was closed. + */ +class IptvRepositoryStalkerPortalStateTest { + + private fun newRepository(): IptvRepository { + val context = io.mockk.mockk(relaxed = true) + val okHttpClient = io.mockk.mockk(relaxed = true) + val profileManager = io.mockk.mockk(relaxed = true) + val invalidationBus = io.mockk.mockk(relaxed = true) + return IptvRepository(context, okHttpClient, profileManager, invalidationBus) + } + + private val portalA = StalkerPortalEntry( + id = "portal-a", + name = "Portal A", + portalUrl = "http://portal-a.invalid/c/", + macAddress = "00:1A:79:00:00:01" + ) + + private val portalB = StalkerPortalEntry( + id = "portal-b", + name = "Portal B", + portalUrl = "http://portal-b.invalid/c/", + macAddress = "00:1A:79:00:00:02" + ) + + /** What a portal that answered returns: its session and its channels. */ + private fun answered(portal: StalkerPortalEntry) = IptvRepository.StalkerPortalChannels( + portalId = portal.id, + api = io.mockk.mockk(relaxed = true), + channels = listOf( + IptvChannel( + id = "stalker:${portal.id}:1", + name = "${portal.name} 1", + streamUrl = "http://${portal.id}.invalid/play/live.php", + group = "News" + ) + ) + ) + + /** What a failed handshake returns: no session, no channels. */ + private fun unreachable(portal: StalkerPortalEntry) = IptvRepository.StalkerPortalChannels( + portalId = portal.id, + api = null, + channels = emptyList() + ) + + @Test + fun `a portal that failed is asked again by the next ordinary load`() = runBlocking { + val repository = newRepository() + val portals = listOf(portalA, portalB) + // The two portals are downloaded in parallel, so the order of attempts + // is not fixed — only how many each portal got. + val attempts = Collections.synchronizedList(mutableListOf()) + var portalBIsUp = false + val fetch: suspend (StalkerPortalEntry) -> IptvRepository.StalkerPortalChannels = { portal -> + attempts += portal.id + if (portal.id == portalB.id && !portalBIsUp) unreachable(portal) else answered(portal) + } + + val partial = repository.loadStalkerChannels(portals, fetchPortal = fetch) + assertThat(partial.second.map { it.id }).containsExactly("stalker:portal-a:1") + assertThat(partial.first.keys).containsExactly("portal-a") + + portalBIsUp = true + val recovered = repository.loadStalkerChannels(portals, fetchPortal = fetch) + + // B is asked again straight away instead of waiting out the window, + // and A is not downloaded a second time to make that possible. + assertThat(attempts.count { it == portalA.id }).isEqualTo(1) + assertThat(attempts.count { it == portalB.id }).isEqualTo(2) + assertThat(recovered.second.map { it.id }) + .containsExactly("stalker:portal-a:1", "stalker:portal-b:1").inOrder() + assertThat(recovered.first.keys).containsExactly("portal-a", "portal-b") + } + + @Test + fun `removing the last portal releases its channel list and session`() = runBlocking { + val repository = newRepository() + val attempts = Collections.synchronizedList(mutableListOf()) + val fetch: suspend (StalkerPortalEntry) -> IptvRepository.StalkerPortalChannels = { portal -> + attempts += portal.id + answered(portal) + } + val playlist = IptvPlaylistEntry( + id = "list-1", + name = "My Provider", + m3uUrl = "http://example.invalid/list.m3u", + enabled = true + ) + + repository.loadStalkerChannels(listOf(portalA), fetchPortal = fetch) + + // An unrelated playlist change runs through the very same path. The + // portal is untouched, so its download has to survive it — dropping it + // here is what cost a second full channel list on every toggle. + repository.ensureCacheOwnership( + "profile-1", + IptvConfig(stalkerPortals = listOf(portalA), playlists = listOf(playlist)) + ) + repository.loadStalkerChannels(listOf(portalA), fetchPortal = fetch) + assertThat(attempts).containsExactly("portal-a") + + // The last portal is removed. Nothing asks for a Stalker channel list + // any more, so this is the only place the cleanup can still happen. + repository.ensureCacheOwnership("profile-1", IptvConfig(playlists = listOf(playlist))) + + // Nothing of the removed portal is left: re-adding it downloads again. + repository.loadStalkerChannels(listOf(portalA), fetchPortal = fetch) + assertThat(attempts).containsExactly("portal-a", "portal-a") + } + + @Test + fun `disabling one of two portals leaves the other one's list alone`() = runBlocking { + val repository = newRepository() + val attempts = Collections.synchronizedList(mutableListOf()) + val fetch: suspend (StalkerPortalEntry) -> IptvRepository.StalkerPortalChannels = { portal -> + attempts += portal.id + answered(portal) + } + + repository.loadStalkerChannels(listOf(portalA, portalB), fetchPortal = fetch) + repository.ensureCacheOwnership( + "profile-1", + IptvConfig(stalkerPortals = listOf(portalA, portalB.copy(enabled = false))) + ) + + repository.loadStalkerChannels(listOf(portalA), fetchPortal = fetch) + assertThat(attempts.count { it == portalA.id }).isEqualTo(1) + + // B is gone from memory; enabling it again really re-downloads it. + repository.loadStalkerChannels(listOf(portalA, portalB), fetchPortal = fetch) + assertThat(attempts.count { it == portalA.id }).isEqualTo(1) + assertThat(attempts.count { it == portalB.id }).isEqualTo(2) + } +} diff --git a/app/src/test/kotlin/com/arflix/tv/data/repository/StalkerChannelListLoaderTest.kt b/app/src/test/kotlin/com/arflix/tv/data/repository/StalkerChannelListLoaderTest.kt new file mode 100644 index 000000000..fdc19a1a2 --- /dev/null +++ b/app/src/test/kotlin/com/arflix/tv/data/repository/StalkerChannelListLoaderTest.kt @@ -0,0 +1,322 @@ +package com.arflix.tv.data.repository + +import com.google.common.truth.Truth.assertThat +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import org.junit.Test + +/** + * The three entry points that ask for the Stalker channel list used to download + * it once each — measured 3 x 29 MB in 17 seconds on a portal with ~21k + * channels. These tests pin down the sharing that turns that into one download. + */ +class StalkerChannelListLoaderTest { + + private var clock = 0L + + private fun TestScope.downloadScope(): CoroutineScope = + CoroutineScope(SupervisorJob() + StandardTestDispatcher(testScheduler)) + + private fun loader( + scope: CoroutineScope, + freshnessWindowMs: Long = 1_000L, + isReusable: (String) -> Boolean = { true } + ) = StalkerChannelListLoader( + freshnessWindowMs = freshnessWindowMs, + scope = scope, + nowMs = { clock }, + isReusable = isReusable + ) + + @Test + fun reusesTheListInsideTheFreshnessWindow() = runTest { + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; "channels" } + + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels") + clock = 999 + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels") + + assertThat(downloads).isEqualTo(1) + scope.cancel() + } + + @Test + fun downloadsAgainOnceTheFreshnessWindowHasPassed() = runTest { + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; "channels" } + + loader.load("portal-a", fetch = fetch) + clock = 1_000 + loader.load("portal-a", fetch = fetch) + + assertThat(downloads).isEqualTo(2) + scope.cancel() + } + + @Test + fun concurrentCallersShareOneDownload() = runTest { + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; delay(2_000); "channels" } + + val startupPrefetch = async { loader.load("portal-a", fetch = fetch) } + val snapshotLoad = async { loader.load("portal-a", fetch = fetch) } + val guideBackfill = async { loader.load("portal-a", fetch = fetch) } + advanceUntilIdle() + + assertThat(startupPrefetch.await()).isEqualTo("channels") + assertThat(snapshotLoad.await()).isEqualTo("channels") + assertThat(guideBackfill.await()).isEqualTo("channels") + assertThat(downloads).isEqualTo(1) + scope.cancel() + } + + @Test + fun aCallerThatGivesUpDoesNotCancelTheSharedDownload() = runTest { + // The startup prefetch times out after 25 s; the snapshot load waiting + // on the same download must still get its channels. + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; delay(5_000); "channels" } + + val givesUp = launch { loader.load("portal-a", fetch = fetch) } + advanceTimeBy(1_000) + givesUp.cancel() + advanceUntilIdle() + + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels") + assertThat(downloads).isEqualTo(1) + scope.cancel() + } + + @Test + fun everyPortalKeepsItsOwnList() = runTest { + // One entry per portal: a second portal neither reuses nor replaces the + // first one's list, and asking the first one again costs nothing. + val scope = downloadScope() + val loader = loader(scope) + val downloaded = mutableListOf() + val fetchA: suspend () -> String = { downloaded += "a"; "channels-a" } + val fetchB: suspend () -> String = { downloaded += "b"; "channels-b" } + + assertThat(loader.load("portal-a", fetch = fetchA)).isEqualTo("channels-a") + assertThat(loader.load("portal-b", fetch = fetchB)).isEqualTo("channels-b") + assertThat(loader.load("portal-a", fetch = fetchA)).isEqualTo("channels-a") + + assertThat(downloaded).containsExactly("a", "b").inOrder() + scope.cancel() + } + + @Test + fun aPortalThatFailedIsRetriedWhileTheOthersKeepTheirLists() = runTest { + // ⭐ Prodigy's review of #677: with two portals configured, one that is + // down must not hide behind one that answered. Its empty answer is not + // remembered, so the next load asks it again — and the healthy portal + // is not downloaded a second time for it. + val scope = downloadScope() + val loader = loader(scope, isReusable = { it.isNotEmpty() }) + val downloaded = mutableListOf() + var portalBIsUp = false + val fetchA: suspend () -> String = { downloaded += "a"; "channels-a" } + val fetchB: suspend () -> String = { + downloaded += "b" + if (portalBIsUp) "channels-b" else "" + } + + loader.load("portal-a", fetch = fetchA) + assertThat(loader.load("portal-b", fetch = fetchB)).isEmpty() + + portalBIsUp = true + loader.load("portal-a", fetch = fetchA) + assertThat(loader.load("portal-b", fetch = fetchB)).isEqualTo("channels-b") + + assertThat(downloaded).containsExactly("a", "b", "b").inOrder() + scope.cancel() + } + + @Test + fun twoForcedCallersAtOnceShareOneDownload() = runTest { + // Measured on device: two entry points reacted to one added playlist and + // pulled the full channel list twice in the same second. Asking for + // fresh data must mean "not the old list", not "a second portal + // session". + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; delay(2_000); "channels" } + + clock = 5_000 + val first = async { loader.load("portal-a", freshSinceMs = 5_000, fetch = fetch) } + val second = async { loader.load("portal-a", freshSinceMs = 5_000, fetch = fetch) } + advanceUntilIdle() + + assertThat(first.await()).isEqualTo("channels") + assertThat(second.await()).isEqualTo("channels") + assertThat(downloads).isEqualTo(1) + scope.cancel() + } + + @Test + fun aForcedCallerTakesTheListThatArrivedWhileItWaitedItsTurn() = runTest { + // ⭐ The bug measured on 09.09.: a playlist toggle wakes two entry + // points, the repository lock serializes them, so the second never + // meets a *running* download — it meets one that just finished. Both + // asked at 5_000, so a list downloaded at 5_400 is newer than the + // request and answers it. Skipping every remembered list instead cost a + // second full 27.67 MB download. + val scope = downloadScope() + val loader = loader(scope, freshnessWindowMs = 300_000L) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; clock = 5_400; "channels" } + + clock = 5_000 + val askedAtMs = clock + assertThat(loader.load("portal-a", freshSinceMs = askedAtMs, fetch = fetch)).isEqualTo("channels") + assertThat(loader.load("portal-a", freshSinceMs = askedAtMs, fetch = fetch)).isEqualTo("channels") + + assertThat(downloads).isEqualTo(1) + scope.cancel() + } + + @Test + fun aForcedCallerDoesNotTakeAListFromBeforeItAsked() = runTest { + // The counterpart: pressing "Refresh IPTV" must reach the portal even + // though the stored list is still well inside the freshness window. + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; "channels-$downloads" } + + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels-1") + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels-1") + clock = 500 + assertThat(loader.load("portal-a", freshSinceMs = 500, fetch = fetch)).isEqualTo("channels-2") + + assertThat(downloads).isEqualTo(2) + scope.cancel() + } + + @Test + fun anEditedPortalIsDownloadedAgainUnderItsNewKey() = runTest { + // The key carries the portal's URL and MAC, so editing either of them + // makes a different portal as far as this loader is concerned. Nothing + // else discards a list: everything the app calls "invalidate" for (a + // playlist, an EPG URL, a profile) leaves the portals alone. + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; "channels" } + + loader.load("portal-a-old-mac", fetch = fetch) + loader.load("portal-a-new-mac", fetch = fetch) + + assertThat(downloads).isEqualTo(2) + scope.cancel() + } + + @Test + fun retainOnlyReleasesTheListsOfPortalsThatAreGone() = runTest { + // Removing a portal means nobody calls load() for it again, so its list + // and session would be held for the rest of the app run. Portals that + // are still configured must not pay for that. + val scope = downloadScope() + val loader = loader(scope) + val downloaded = mutableListOf() + val fetchA: suspend () -> String = { downloaded += "a"; "channels-a" } + val fetchB: suspend () -> String = { downloaded += "b"; "channels-b" } + + loader.load("portal-a", fetch = fetchA) + loader.load("portal-b", fetch = fetchB) + + // Portal B removed, A still configured: only B is dropped. + loader.retainOnly(setOf("portal-a")) + loader.load("portal-a", fetch = fetchA) + loader.load("portal-b", fetch = fetchB) + assertThat(downloaded).containsExactly("a", "b", "b").inOrder() + + // The last portal removed: everything goes. + loader.retainOnly(emptySet()) + loader.load("portal-a", fetch = fetchA) + assertThat(downloaded).containsExactly("a", "b", "b", "a").inOrder() + scope.cancel() + } + + @Test + fun aDetachedDownloadCannotOverwriteTheOneThatReplacedIt() = runTest { + // The portal is removed and re-added while a download is still running + // — a portal edited back to its old URL and MAC does exactly this. The + // detached download finishes last here; its result must not land. + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { + val attempt = ++downloads + // The first download is the slow one, so it lands after the second. + delay(if (attempt == 1) 4_000 else 1_000) + "channels-$attempt" + } + + val detached = async { loader.load("portal-a", fetch = fetch) } + advanceTimeBy(500) + loader.retainOnly(emptySet()) + val replacement = async { loader.load("portal-a", fetch = fetch) } + advanceUntilIdle() + + assertThat(detached.await()).isEqualTo("channels-1") + assertThat(replacement.await()).isEqualTo("channels-2") + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels-2") + assertThat(downloads).isEqualTo(2) + scope.cancel() + } + + @Test + fun anEmptyResultIsNotRemembered() = runTest { + // A failed handshake returns an empty list; remembering it would keep + // the portal dark for the whole freshness window. + val scope = downloadScope() + val loader = loader(scope, isReusable = { it.isNotEmpty() }) + var downloads = 0 + val fetch: suspend () -> String = { downloads++; if (downloads == 1) "" else "channels" } + + assertThat(loader.load("portal-a", fetch = fetch)).isEmpty() + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels") + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels") + + assertThat(downloads).isEqualTo(2) + scope.cancel() + } + + @Test + fun aFailedDownloadReachesTheCallerAndIsNotRemembered() = runTest { + val scope = downloadScope() + val loader = loader(scope) + var downloads = 0 + val fetch: suspend () -> String = { + downloads++ + if (downloads == 1) throw IllegalStateException("portal unreachable") else "channels" + } + + val failure = runCatching { loader.load("portal-a", fetch = fetch) }.exceptionOrNull() + assertThat(failure).isInstanceOf(IllegalStateException::class.java) + assertThat(loader.load("portal-a", fetch = fetch)).isEqualTo("channels") + + assertThat(downloads).isEqualTo(2) + scope.cancel() + } +}