feat(tmdb): deduplicate concurrent in-flight API requests
Add CompletableDeferred-based request coalescing to TmdbService and TmdbMetadataService. When multiple coroutines request the same TMDB lookup simultaneously, only the first one performs the actual network call while subsequent callers await its result. - Add enrichmentInFlight/episodeInFlight maps to TmdbMetadataService - Add imdbToTmdbInFlight/tmdbToImdbInFlight maps to TmdbService - Handle CancellationException separately to propagate cancellation - Clean up in-flight entries in finally blocks - Add unit test verifying deduplication behavior
This commit is contained in:
HALİL İBRAHİM ÖZDEMİR
parent
2d287989de
commit
2afa7f1155
3 changed files with 145 additions and 19 deletions
|
|
@ -15,6 +15,8 @@ import com.nuvio.tv.domain.model.MetaCompany
|
|||
import com.nuvio.tv.domain.model.MetaPreview
|
||||
import com.nuvio.tv.domain.model.PersonDetail
|
||||
import com.nuvio.tv.domain.model.PosterShape
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
|
|
@ -36,6 +38,8 @@ class TmdbMetadataService @Inject constructor(
|
|||
// In-memory caches
|
||||
private val enrichmentCache = ConcurrentHashMap<String, TmdbEnrichment>()
|
||||
private val episodeCache = ConcurrentHashMap<String, Map<Pair<Int, Int>, TmdbEpisodeEnrichment>>()
|
||||
private val enrichmentInFlight = ConcurrentHashMap<String, CompletableDeferred<TmdbEnrichment?>>()
|
||||
private val episodeInFlight = ConcurrentHashMap<String, CompletableDeferred<Map<Pair<Int, Int>, TmdbEpisodeEnrichment>>>()
|
||||
private val personCache = ConcurrentHashMap<String, PersonDetail>()
|
||||
private val moreLikeThisCache = ConcurrentHashMap<String, List<MetaPreview>>()
|
||||
private val entityHeaderCache = ConcurrentHashMap<String, TmdbEntityHeader>()
|
||||
|
|
@ -51,8 +55,13 @@ class TmdbMetadataService @Inject constructor(
|
|||
val normalizedLanguage = normalizeTmdbLanguage(language)
|
||||
val cacheKey = "$tmdbId:${contentType.name}:$normalizedLanguage"
|
||||
enrichmentCache[cacheKey]?.let { return@withContext it }
|
||||
enrichmentInFlight[cacheKey]?.let { return@withContext it.await() }
|
||||
|
||||
val numericId = tmdbId.toIntOrNull() ?: return@withContext null
|
||||
val requestDeferred = CompletableDeferred<TmdbEnrichment?>()
|
||||
enrichmentInFlight.putIfAbsent(cacheKey, requestDeferred)?.let { existing ->
|
||||
return@withContext existing.await()
|
||||
}
|
||||
val tmdbType = when (contentType) {
|
||||
ContentType.SERIES, ContentType.TV -> "tv"
|
||||
else -> "movie"
|
||||
|
|
@ -294,10 +303,17 @@ class TmdbMetadataService @Inject constructor(
|
|||
collectionName = collectionName
|
||||
)
|
||||
enrichmentCache[cacheKey] = enrichment
|
||||
requestDeferred.complete(enrichment)
|
||||
enrichment
|
||||
} catch (e: CancellationException) {
|
||||
requestDeferred.cancel(e)
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
Log.e(TAG, "Failed to fetch TMDB enrichment: ${e.message}", e)
|
||||
requestDeferred.complete(null)
|
||||
null
|
||||
} finally {
|
||||
enrichmentInFlight.remove(cacheKey, requestDeferred)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -309,27 +325,41 @@ class TmdbMetadataService @Inject constructor(
|
|||
val normalizedLanguage = normalizeTmdbLanguage(language)
|
||||
val cacheKey = "$tmdbId:${seasonNumbers.sorted().joinToString(",")}:$normalizedLanguage"
|
||||
episodeCache[cacheKey]?.let { return@withContext it }
|
||||
episodeInFlight[cacheKey]?.let { return@withContext it.await() }
|
||||
|
||||
val numericId = tmdbId.toIntOrNull() ?: return@withContext emptyMap()
|
||||
val requestDeferred = CompletableDeferred<Map<Pair<Int, Int>, TmdbEpisodeEnrichment>>()
|
||||
episodeInFlight.putIfAbsent(cacheKey, requestDeferred)?.let { existing ->
|
||||
return@withContext existing.await()
|
||||
}
|
||||
val result = mutableMapOf<Pair<Int, Int>, TmdbEpisodeEnrichment>()
|
||||
|
||||
seasonNumbers.distinct().forEach { season ->
|
||||
try {
|
||||
val response = tmdbApi.getTvSeasonDetails(numericId, season, TMDB_API_KEY, normalizedLanguage)
|
||||
val episodes = response.body()?.episodes.orEmpty()
|
||||
episodes.forEach { ep ->
|
||||
val epNum = ep.episodeNumber ?: return@forEach
|
||||
result[season to epNum] = ep.toEnrichment()
|
||||
try {
|
||||
seasonNumbers.distinct().forEach { season ->
|
||||
try {
|
||||
val response = tmdbApi.getTvSeasonDetails(numericId, season, TMDB_API_KEY, normalizedLanguage)
|
||||
val episodes = response.body()?.episodes.orEmpty()
|
||||
episodes.forEach { ep ->
|
||||
val epNum = ep.episodeNumber ?: return@forEach
|
||||
result[season to epNum] = ep.toEnrichment()
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
Log.w(TAG, "Failed to fetch TMDB season $season: ${e.message}")
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
Log.w(TAG, "Failed to fetch TMDB season $season: ${e.message}")
|
||||
}
|
||||
}
|
||||
|
||||
if (result.isNotEmpty()) {
|
||||
episodeCache[cacheKey] = result
|
||||
val finalResult = result.toMap()
|
||||
if (finalResult.isNotEmpty()) {
|
||||
episodeCache[cacheKey] = finalResult
|
||||
}
|
||||
requestDeferred.complete(finalResult)
|
||||
finalResult
|
||||
} catch (e: CancellationException) {
|
||||
requestDeferred.cancel(e)
|
||||
throw e
|
||||
} finally {
|
||||
episodeInFlight.remove(cacheKey, requestDeferred)
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
suspend fun fetchMoreLikeThis(
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ package com.nuvio.tv.core.tmdb
|
|||
import android.util.Log
|
||||
import com.nuvio.tv.BuildConfig
|
||||
import com.nuvio.tv.data.remote.api.TmdbApi
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
|
|
@ -27,6 +29,9 @@ class TmdbService @Inject constructor(
|
|||
|
||||
// Cache: TMDB ID -> IMDB ID
|
||||
private val tmdbToImdbCache = ConcurrentHashMap<Int, String>()
|
||||
|
||||
private val imdbToTmdbInFlight = ConcurrentHashMap<String, CompletableDeferred<Int?>>()
|
||||
private val tmdbToImdbInFlight = ConcurrentHashMap<String, CompletableDeferred<String?>>()
|
||||
|
||||
// Mutex for thread-safe cache operations
|
||||
private val cacheMutex = Mutex()
|
||||
|
|
@ -51,6 +56,13 @@ class TmdbService @Inject constructor(
|
|||
return@withContext cached
|
||||
}
|
||||
|
||||
val normalizedType = normalizeMediaType(mediaType)
|
||||
val requestKey = "$imdbId:$normalizedType"
|
||||
val requestDeferred = CompletableDeferred<Int?>()
|
||||
imdbToTmdbInFlight.putIfAbsent(requestKey, requestDeferred)?.let { existing ->
|
||||
return@withContext existing.await()
|
||||
}
|
||||
|
||||
try {
|
||||
Log.d(TAG, "Looking up TMDB ID for IMDB: $imdbId (type: $mediaType)")
|
||||
|
||||
|
|
@ -62,13 +74,17 @@ class TmdbService @Inject constructor(
|
|||
|
||||
if (!response.isSuccessful) {
|
||||
Log.e(TAG, "TMDB API error: ${response.code()} - ${response.message()}")
|
||||
requestDeferred.complete(null)
|
||||
return@withContext null
|
||||
}
|
||||
|
||||
val body = response.body() ?: return@withContext null
|
||||
val body = response.body()
|
||||
if (body == null) {
|
||||
requestDeferred.complete(null)
|
||||
return@withContext null
|
||||
}
|
||||
|
||||
// Determine which results to use based on media type
|
||||
val normalizedType = normalizeMediaType(mediaType)
|
||||
val result = when (normalizedType) {
|
||||
"movie" -> body.movieResults?.firstOrNull()
|
||||
"tv", "series" -> body.tvResults?.firstOrNull()
|
||||
|
|
@ -83,16 +99,25 @@ class TmdbService @Inject constructor(
|
|||
imdbToTmdbCache[imdbId] = found.id
|
||||
tmdbToImdbCache[found.id] = imdbId
|
||||
}
|
||||
|
||||
|
||||
requestDeferred.complete(found.id)
|
||||
|
||||
return@withContext found.id
|
||||
}
|
||||
|
||||
Log.w(TAG, "No TMDB result found for IMDB: $imdbId")
|
||||
requestDeferred.complete(null)
|
||||
null
|
||||
|
||||
} catch (e: CancellationException) {
|
||||
requestDeferred.cancel(e)
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
Log.e(TAG, "Error looking up TMDB ID for $imdbId: ${e.message}", e)
|
||||
requestDeferred.complete(null)
|
||||
null
|
||||
} finally {
|
||||
imdbToTmdbInFlight.remove(requestKey, requestDeferred)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -110,10 +135,16 @@ class TmdbService @Inject constructor(
|
|||
return@withContext cached
|
||||
}
|
||||
|
||||
val normalizedType = normalizeMediaType(mediaType)
|
||||
val requestKey = "$tmdbId:$normalizedType"
|
||||
val requestDeferred = CompletableDeferred<String?>()
|
||||
tmdbToImdbInFlight.putIfAbsent(requestKey, requestDeferred)?.let { existing ->
|
||||
return@withContext existing.await()
|
||||
}
|
||||
|
||||
try {
|
||||
Log.d(TAG, "Looking up IMDB ID for TMDB: $tmdbId (type: $mediaType)")
|
||||
|
||||
val normalizedType = normalizeMediaType(mediaType)
|
||||
val response = when (normalizedType) {
|
||||
"movie" -> tmdbApi.getMovieExternalIds(tmdbId, TMDB_API_KEY)
|
||||
"tv", "series" -> tmdbApi.getTvExternalIds(tmdbId, TMDB_API_KEY)
|
||||
|
|
@ -122,10 +153,15 @@ class TmdbService @Inject constructor(
|
|||
|
||||
if (!response.isSuccessful) {
|
||||
Log.e(TAG, "TMDB API error: ${response.code()} - ${response.message()}")
|
||||
requestDeferred.complete(null)
|
||||
return@withContext null
|
||||
}
|
||||
|
||||
val body = response.body() ?: return@withContext null
|
||||
val body = response.body()
|
||||
if (body == null) {
|
||||
requestDeferred.complete(null)
|
||||
return@withContext null
|
||||
}
|
||||
|
||||
body.imdbId?.let { imdbId ->
|
||||
Log.d(TAG, "Found IMDB ID: $imdbId for TMDB: $tmdbId")
|
||||
|
|
@ -135,16 +171,25 @@ class TmdbService @Inject constructor(
|
|||
tmdbToImdbCache[tmdbId] = imdbId
|
||||
imdbToTmdbCache[imdbId] = tmdbId
|
||||
}
|
||||
|
||||
|
||||
requestDeferred.complete(imdbId)
|
||||
|
||||
return@withContext imdbId
|
||||
}
|
||||
|
||||
Log.w(TAG, "No IMDB ID found for TMDB: $tmdbId")
|
||||
requestDeferred.complete(null)
|
||||
null
|
||||
|
||||
} catch (e: CancellationException) {
|
||||
requestDeferred.cancel(e)
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
Log.e(TAG, "Error looking up IMDB ID for $tmdbId: ${e.message}", e)
|
||||
requestDeferred.complete(null)
|
||||
null
|
||||
} finally {
|
||||
tmdbToImdbInFlight.remove(requestKey, requestDeferred)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -203,6 +248,8 @@ class TmdbService @Inject constructor(
|
|||
fun clearCache() {
|
||||
imdbToTmdbCache.clear()
|
||||
tmdbToImdbCache.clear()
|
||||
imdbToTmdbInFlight.clear()
|
||||
tmdbToImdbInFlight.clear()
|
||||
Log.d(TAG, "Cache cleared")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -14,8 +14,12 @@ import com.nuvio.tv.data.remote.api.TmdbNetworkDetailsResponse
|
|||
import com.nuvio.tv.domain.model.ContentType
|
||||
import io.mockk.coEvery
|
||||
import io.mockk.mockk
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.awaitAll
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.coroutines.yield
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNotNull
|
||||
import org.junit.Assert.assertNull
|
||||
|
|
@ -57,6 +61,51 @@ class TmdbMetadataServiceTest {
|
|||
assertEquals(77, enrichment?.networks?.firstOrNull()?.tmdbId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `fetchEnrichment deduplicates concurrent requests for same key`() = runTest {
|
||||
val api = mockk<TmdbApi>()
|
||||
val gate = CompletableDeferred<Unit>()
|
||||
var detailsCalls = 0
|
||||
var creditsCalls = 0
|
||||
var imagesCalls = 0
|
||||
var releaseCalls = 0
|
||||
|
||||
coEvery { api.getMovieDetails(any(), any(), any()) } coAnswers {
|
||||
detailsCalls += 1
|
||||
gate.await()
|
||||
Response.success(TmdbDetailsResponse(id = 10, title = "Movie"))
|
||||
}
|
||||
coEvery { api.getMovieCredits(any(), any(), any()) } coAnswers {
|
||||
creditsCalls += 1
|
||||
Response.success(TmdbCreditsResponse())
|
||||
}
|
||||
coEvery { api.getMovieImages(any(), any(), any()) } coAnswers {
|
||||
imagesCalls += 1
|
||||
Response.success(TmdbImagesResponse())
|
||||
}
|
||||
coEvery { api.getMovieReleaseDates(any(), any()) } coAnswers {
|
||||
releaseCalls += 1
|
||||
Response.success(TmdbMovieReleaseDatesResponse())
|
||||
}
|
||||
|
||||
val service = TmdbMetadataService(api)
|
||||
|
||||
val first = async { service.fetchEnrichment(tmdbId = "10", contentType = ContentType.MOVIE, language = "en") }
|
||||
val second = async { service.fetchEnrichment(tmdbId = "10", contentType = ContentType.MOVIE, language = "en") }
|
||||
|
||||
yield()
|
||||
assertEquals(1, detailsCalls)
|
||||
|
||||
gate.complete(Unit)
|
||||
val results = awaitAll(first, second)
|
||||
|
||||
assertEquals(1, detailsCalls)
|
||||
assertEquals(1, creditsCalls)
|
||||
assertEquals(1, imagesCalls)
|
||||
assertEquals(1, releaseCalls)
|
||||
assertEquals(results[0], results[1])
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `company browse requests movie then tv rails when source type is movie`() = runTest {
|
||||
val api = mockk<TmdbApi>()
|
||||
|
|
|
|||
Loading…
Reference in a new issue