diff --git a/app/src/main/java/com/fluxa/app/core/rust/FluxaHeadlessAppRuntime.kt b/app/src/main/java/com/fluxa/app/core/rust/FluxaHeadlessAppRuntime.kt index 02afb2a..a5c1002 100644 --- a/app/src/main/java/com/fluxa/app/core/rust/FluxaHeadlessAppRuntime.kt +++ b/app/src/main/java/com/fluxa/app/core/rust/FluxaHeadlessAppRuntime.kt @@ -4,8 +4,6 @@ import java.io.Closeable import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock /** * Android-side owner for the Rust headless app runtime. @@ -19,19 +17,18 @@ class FluxaHeadlessAppRuntime( private val environment: HeadlessPlatformEnvironment, private val maxEffectsPerDispatch: Int = 64 ) : Closeable { - private val mutex = Mutex() private val runner = FluxaHeadlessEffectRunner(engine, environment, maxEffectsPerDispatch) private val _state = MutableStateFlow>(emptyMap()) val state: StateFlow> = _state.asStateFlow() suspend fun dispatch(action: Any): NativeHeadlessEngineResult { - return mutex.withLock { runner.dispatchAndDrain(action) }.also { result -> + return runner.dispatchAndDrain(action).also { result -> _state.value = result.state } } suspend fun complete(result: HeadlessEffectCompletion): NativeHeadlessEngineResult { - return mutex.withLock { runner.completeAndDrain(result) }.also { next -> + return runner.completeAndDrain(result).also { next -> _state.value = next.state } } diff --git a/app/src/main/java/com/fluxa/app/di/NetworkModule.kt b/app/src/main/java/com/fluxa/app/di/NetworkModule.kt index 7eb5b31..ad350d1 100644 --- a/app/src/main/java/com/fluxa/app/di/NetworkModule.kt +++ b/app/src/main/java/com/fluxa/app/di/NetworkModule.kt @@ -39,7 +39,72 @@ object NetworkModule { @Singleton fun provideLoggingInterceptor(): HttpLoggingInterceptor { return HttpLoggingInterceptor().apply { - level = HttpLoggingInterceptor.Level.NONE + level = if (BuildConfig.DEBUG) HttpLoggingInterceptor.Level.BASIC else HttpLoggingInterceptor.Level.NONE + } + } + + @Provides + @Singleton + fun provideNetTimingEventListenerFactory(): okhttp3.EventListener.Factory { + return okhttp3.EventListener.Factory { call -> + if (!BuildConfig.DEBUG) return@Factory okhttp3.EventListener.NONE + object : okhttp3.EventListener() { + var callStartNs = 0L + fun elapsedMs() = (System.nanoTime() - callStartNs) / 1_000_000 + + override fun callStart(call: okhttp3.Call) { + callStartNs = System.nanoTime() + android.util.Log.d("NetTiming", "callStart url=${call.request().url}") + } + + override fun dnsStart(call: okhttp3.Call, domainName: String) { + android.util.Log.d("NetTiming", "dnsStart +${elapsedMs()}ms host=$domainName") + } + + override fun dnsEnd(call: okhttp3.Call, domainName: String, inetAddressList: List) { + android.util.Log.d("NetTiming", "dnsEnd +${elapsedMs()}ms host=$domainName resolved=$inetAddressList") + } + + override fun connectStart(call: okhttp3.Call, inetSocketAddress: java.net.InetSocketAddress, proxy: java.net.Proxy) { + android.util.Log.d("NetTiming", "connectStart +${elapsedMs()}ms addr=$inetSocketAddress") + } + + override fun secureConnectStart(call: okhttp3.Call) { + android.util.Log.d("NetTiming", "secureConnectStart +${elapsedMs()}ms") + } + + override fun secureConnectEnd(call: okhttp3.Call, handshake: okhttp3.Handshake?) { + android.util.Log.d("NetTiming", "secureConnectEnd +${elapsedMs()}ms") + } + + override fun connectEnd(call: okhttp3.Call, inetSocketAddress: java.net.InetSocketAddress, proxy: java.net.Proxy, protocol: okhttp3.Protocol?) { + android.util.Log.d("NetTiming", "connectEnd +${elapsedMs()}ms") + } + + override fun connectFailed(call: okhttp3.Call, inetSocketAddress: java.net.InetSocketAddress, proxy: java.net.Proxy, protocol: okhttp3.Protocol?, ioe: java.io.IOException) { + android.util.Log.d("NetTiming", "connectFailed +${elapsedMs()}ms addr=$inetSocketAddress error=$ioe") + } + + override fun requestHeadersEnd(call: okhttp3.Call, request: okhttp3.Request) { + android.util.Log.d("NetTiming", "requestHeadersEnd +${elapsedMs()}ms") + } + + override fun responseHeadersStart(call: okhttp3.Call) { + android.util.Log.d("NetTiming", "responseHeadersStart +${elapsedMs()}ms (time to first byte)") + } + + override fun responseBodyEnd(call: okhttp3.Call, byteCount: Long) { + android.util.Log.d("NetTiming", "responseBodyEnd +${elapsedMs()}ms bytes=$byteCount") + } + + override fun callEnd(call: okhttp3.Call) { + android.util.Log.d("NetTiming", "callEnd +${elapsedMs()}ms url=${call.request().url}") + } + + override fun callFailed(call: okhttp3.Call, ioe: java.io.IOException) { + android.util.Log.d("NetTiming", "callFailed +${elapsedMs()}ms url=${call.request().url} error=$ioe") + } + } } } @@ -54,17 +119,27 @@ object NetworkModule { @Provides @Singleton fun provideDispatcher(): okhttp3.Dispatcher = okhttp3.Dispatcher().apply { - maxRequests = 64 - maxRequestsPerHost = 16 + maxRequests = 96 + maxRequestsPerHost = 32 } @Provides @Singleton @Named("StremioClient") - fun provideStremioOkHttpClient(logging: HttpLoggingInterceptor, connectionPool: okhttp3.ConnectionPool, dispatcher: okhttp3.Dispatcher): OkHttpClient { + fun provideStremioOkHttpClient( + logging: HttpLoggingInterceptor, + connectionPool: okhttp3.ConnectionPool, + dispatcher: okhttp3.Dispatcher, + netTimingEventListenerFactory: okhttp3.EventListener.Factory + ): OkHttpClient { val ipv4OnlyDns = object : okhttp3.Dns { override fun lookup(hostname: String): List { - return okhttp3.Dns.SYSTEM.lookup(hostname).filter { it is java.net.Inet4Address } + val all = okhttp3.Dns.SYSTEM.lookup(hostname) + val filtered = all.filter { it is java.net.Inet4Address } + if (BuildConfig.DEBUG) { + android.util.Log.d("NetTiming", "dns host=$hostname all=$all filteredToIpv4=$filtered") + } + return filtered } } @@ -72,6 +147,7 @@ object NetworkModule { .dns(ipv4OnlyDns) .connectionPool(connectionPool) .dispatcher(dispatcher) + .eventListenerFactory(netTimingEventListenerFactory) .addInterceptor { chain -> chain.proceed(HttpRequestSecurity.upgradeRemoteHttpRequest(chain.request())) } diff --git a/app/src/main/java/com/fluxa/app/ui/catalog/AndroidDiscoverDataSource.kt b/app/src/main/java/com/fluxa/app/ui/catalog/AndroidDiscoverDataSource.kt index c3ef27f..244bdc4 100644 --- a/app/src/main/java/com/fluxa/app/ui/catalog/AndroidDiscoverDataSource.kt +++ b/app/src/main/java/com/fluxa/app/ui/catalog/AndroidDiscoverDataSource.kt @@ -10,9 +10,11 @@ import com.fluxa.app.shared.feature.discover.DiscoverFilterOptionUiModel import com.fluxa.app.shared.feature.discover.DiscoverFiltersUiModel import com.fluxa.app.shared.feature.discover.DiscoverUiState import com.fluxa.app.domain.discovery.DiscoverCatalogOption +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.flowOn class AndroidDiscoverDataSource( private val homeViewModel: HomeViewModel, @@ -61,15 +63,21 @@ class AndroidDiscoverDataSource( }, isLoading = state.isLoading ) - } + }.flowOn(Dispatchers.Default) override suspend fun updateFilters(filters: DiscoverFiltersUiModel) { val previousFilters = this.filters.value if (filters != previousFilters) homeViewModel.clearDiscoverResults() homeViewModel.setDiscoverLoading(true) - val availableCatalogs = homeViewModel.discoverCatalogOptions(filters.contentType) - catalogOptions.value = availableCatalogs - contentTypes.value = homeViewModel.discoverContentTypes() + val contentTypeUnchanged = filters.contentType == previousFilters.contentType + val availableCatalogs = if (contentTypeUnchanged && catalogOptions.value.isNotEmpty()) { + catalogOptions.value + } else { + homeViewModel.discoverCatalogOptions(filters.contentType).also { catalogOptions.value = it } + } + if (!contentTypeUnchanged || contentTypes.value.isEmpty()) { + contentTypes.value = homeViewModel.discoverContentTypes() + } val selectedCatalogKey = filters.catalogKey ?.takeIf { key -> availableCatalogs.any { it.key == key } } ?: availableCatalogs.firstOrNull()?.key diff --git a/app/src/main/java/com/fluxa/app/ui/catalog/HomeBillboardRuntime.kt b/app/src/main/java/com/fluxa/app/ui/catalog/HomeBillboardRuntime.kt index 58a33b3..1e2fa84 100644 --- a/app/src/main/java/com/fluxa/app/ui/catalog/HomeBillboardRuntime.kt +++ b/app/src/main/java/com/fluxa/app/ui/catalog/HomeBillboardRuntime.kt @@ -74,6 +74,12 @@ internal class HomeBillboardRuntime( rotationJob?.cancel() } + fun pauseBackgroundWork() { + rotationJob?.cancel() + prefetchJob?.cancel() + trailerJob?.cancel() + } + fun next() { pauseRotation() val items = pool() diff --git a/app/src/main/java/com/fluxa/app/ui/catalog/HomeCatalogFeedCoordinator.kt b/app/src/main/java/com/fluxa/app/ui/catalog/HomeCatalogFeedCoordinator.kt index e2de040..f6476cf 100644 --- a/app/src/main/java/com/fluxa/app/ui/catalog/HomeCatalogFeedCoordinator.kt +++ b/app/src/main/java/com/fluxa/app/ui/catalog/HomeCatalogFeedCoordinator.kt @@ -16,10 +16,13 @@ import com.fluxa.app.domain.discovery.metadataFeedHomeTitle import com.fluxa.app.domain.discovery.orderedMetadataFeeds import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock internal class HomeCatalogFeedCoordinator( private val repository: StremioRepository, @@ -33,6 +36,8 @@ internal class HomeCatalogFeedCoordinator( private val setCategories: (List) -> Unit, private val currentCategories: () -> List ) { + private var remainingCatalogsJob: Job? = null + suspend fun buildInitialCategories(profile: UserProfile?): List { val lang = profile?.safeLanguage ?: "en" val categories = mutableListOf() @@ -272,9 +277,10 @@ internal class HomeCatalogFeedCoordinator( } fun loadRemainingCatalogs(profile: UserProfile?) { + remainingCatalogsJob?.cancel() val lang = profile?.safeLanguage ?: "en" - scope.launch(Dispatchers.IO) { + remainingCatalogsJob = scope.launch(Dispatchers.IO) { try { val enabledFeeds = getMetadataFeeds(profile) .let { orderedMetadataFeeds(it, profile?.homeFeedOrder) } @@ -284,21 +290,27 @@ internal class HomeCatalogFeedCoordinator( feeds.filter { isMetadataFeedEnabled(selectedKeys, it.key) } } .drop(2) - val firstWave = enabledFeeds.take(8).map { feed -> - async { fetchAddonFeedCategory(feed, lang) } - } - setCategories(optimizeHomeCategories(currentCategories() + firstWave.awaitAll().filterNotNull(), lang)) - val secondWave = enabledFeeds.drop(8).map { feed -> - async { fetchAddonFeedCategory(feed, lang) } - } - setCategories(optimizeHomeCategories(currentCategories() + secondWave.awaitAll().filterNotNull(), lang)) + val publishMutex = Mutex() + enabledFeeds.map { feed -> + async { + val category = fetchAddonFeedCategory(feed, lang) ?: return@async + publishMutex.withLock { + setCategories(optimizeHomeCategories(currentCategories() + category, lang)) + } + } + }.awaitAll() } catch (e: Exception) { Log.e("HomeViewModel", "Remaining catalogs failed", e) } } } + fun pauseRemainingCatalogs() { + remainingCatalogsJob?.cancel() + remainingCatalogsJob = null + } + suspend fun getMetadataFeeds(profile: UserProfile?): List { val addons = userAddons().ifEmpty { addonRepository.getUserAddons(profile?.authKey ?: "", profile?.safeLocalAddons) diff --git a/app/src/main/java/com/fluxa/app/ui/catalog/HomeViewModel.kt b/app/src/main/java/com/fluxa/app/ui/catalog/HomeViewModel.kt index 1659bc3..fb45f8c 100644 --- a/app/src/main/java/com/fluxa/app/ui/catalog/HomeViewModel.kt +++ b/app/src/main/java/com/fluxa/app/ui/catalog/HomeViewModel.kt @@ -906,7 +906,13 @@ class HomeViewModel @Inject constructor( browseCoordinator.setLoading(isLoading) } + fun pauseHomeBackgroundWork() { + billboardRuntime.pauseBackgroundWork() + feedCoordinator.pauseRemainingCatalogs() + } + fun discover(type: String, catalogKey: String?, genre: String?, year: String?, rating: Float?, provider: String?, region: String?) { + pauseHomeBackgroundWork() browseCoordinator.discover(type, catalogKey, genre, year, rating, provider, region) } diff --git a/core/src/commonMain/kotlin/com/fluxa/app/core/rust/FluxaHeadlessEffectRunner.kt b/core/src/commonMain/kotlin/com/fluxa/app/core/rust/FluxaHeadlessEffectRunner.kt index 97772d0..36e8532 100644 --- a/core/src/commonMain/kotlin/com/fluxa/app/core/rust/FluxaHeadlessEffectRunner.kt +++ b/core/src/commonMain/kotlin/com/fluxa/app/core/rust/FluxaHeadlessEffectRunner.kt @@ -1,5 +1,8 @@ package com.fluxa.app.core.rust +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock + class NativeHeadlessEngineResult( val effects: List = emptyList(), stateProvider: () -> Map = { emptyMap() } @@ -17,10 +20,13 @@ class FluxaHeadlessEffectRunner( private val environment: HeadlessPlatformEnvironment, private val maxEffectsPerDispatch: Int = 64 ) { - suspend fun dispatchAndDrain(action: Any): NativeHeadlessEngineResult = drain(engine.dispatch(action)) + private val engineMutex = Mutex() + + suspend fun dispatchAndDrain(action: Any): NativeHeadlessEngineResult = + drain(engineMutex.withLock { engine.dispatch(action) }) suspend fun completeAndDrain(result: HeadlessEffectCompletion): NativeHeadlessEngineResult { - return drain(engine.completeEffect(result)) + return drain(engineMutex.withLock { engine.completeEffect(result) }) } private suspend fun drain(initial: NativeHeadlessEngineResult): NativeHeadlessEngineResult { @@ -33,7 +39,8 @@ class FluxaHeadlessEffectRunner( } var remaining = maxEffectsPerDispatch while (pending.isNotEmpty() && remaining > 0) { - current = engine.completeEffect(environment.execute(pending.removeFirst())) + val executed = environment.execute(pending.removeFirst()) + current = engineMutex.withLock { engine.completeEffect(executed) } patches += current current.effects.forEach { effect -> if (queuedIds.add(effect.id)) pending.addLast(effect) diff --git a/data/src/androidMain/kotlin/com/fluxa/app/data/repository/AddonRepository.kt b/data/src/androidMain/kotlin/com/fluxa/app/data/repository/AddonRepository.kt index c1406a2..e4dcf72 100644 --- a/data/src/androidMain/kotlin/com/fluxa/app/data/repository/AddonRepository.kt +++ b/data/src/androidMain/kotlin/com/fluxa/app/data/repository/AddonRepository.kt @@ -3,22 +3,34 @@ package com.fluxa.app.data.repository import android.util.Log import com.fluxa.app.data.remote.* import com.fluxa.app.domain.discovery.supportsStremioResource +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Deferred import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.selects.select import kotlinx.coroutines.supervisorScope import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeoutOrNull +import java.util.concurrent.ConcurrentHashMap import javax.inject.Inject import javax.inject.Singleton private const val ADDON_META_TIMEOUT_MS = 3000L +private const val ADDON_META_HEDGE_DELAY_MS = 500L +private const val META_DETAIL_CACHE_TTL_MS = 5 * 60 * 1000L @Singleton class AddonRepository @Inject constructor( private val addonManifestClient: StremioAddonManifestClient, private val addonResourceClient: StremioAddonResourceClient ) { + private val metaDetailRepositoryScope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + private val metaDetailCache = ConcurrentHashMap>() + private val metaDetailInFlight = ConcurrentHashMap>() + suspend fun getAddonManifest( transportUrl: String, forceRefresh: Boolean = false @@ -29,43 +41,84 @@ class AddonRepository @Inject constructor( id: String, authKey: String, localAddons: List? = emptyList() + ): MetaDetail? { + val cacheKey = "$type:$id" + metaDetailCache[cacheKey]?.let { (cachedAt, value) -> + if (System.currentTimeMillis() - cachedAt < META_DETAIL_CACHE_TTL_MS) return value + } + + val deferred = metaDetailInFlight.computeIfAbsent(cacheKey) { + metaDetailRepositoryScope.async { + try { + fetchAddonMetaDetailUncached(type, id, authKey, localAddons) + } finally { + metaDetailInFlight.remove(cacheKey) + } + } + } + val result = deferred.await() + metaDetailCache[cacheKey] = System.currentTimeMillis() to result + return result + } + + private suspend fun fetchAddonMetaDetailUncached( + type: String, + id: String, + authKey: String, + localAddons: List? = emptyList() ): MetaDetail? { val allAddons = getUserAddons(authKey, localAddons) val addons = allAddons.filter { it.supportsStremioResource("meta") } Log.d("MetaFetch", "getAddonMetaDetail type=$type id=${id.take(30)}: total=${allAddons.size} supported=${addons.size} names=${addons.map { it.manifest.name }}") if (addons.isEmpty()) return null - val results = supervisorScope { - addons.map { addon -> - async { - withTimeoutOrNull(ADDON_META_TIMEOUT_MS) { - addonResourceClient.getMetaDetailFromAddonResult(addon.transportUrl, type, id) - } ?: AddonResourceResult.NetworkError( - url = addon.transportUrl, - cause = null + return coroutineScope { + fun launchLookup(transportUrl: String) = async { + withTimeoutOrNull(ADDON_META_TIMEOUT_MS) { + addonResourceClient.getMetaDetailFromAddonResult(transportUrl, type, id) + } ?: AddonResourceResult.NetworkError(url = transportUrl, cause = null) + } + + val primaryDeferred = launchLookup(addons.first().transportUrl) + val hedgedResult = withTimeoutOrNull(ADDON_META_HEDGE_DELAY_MS) { primaryDeferred.await() } + if (hedgedResult is AddonResourceResult.Success) { + return@coroutineScope hedgedResult.value + } + + val remainingAddons = addons.drop(1) + if (remainingAddons.isEmpty()) { + val result = if (hedgedResult != null) hedgedResult else primaryDeferred.await() + return@coroutineScope (result as? AddonResourceResult.Success)?.value + } + + val pending = mutableListOf(primaryDeferred) + pending += remainingAddons.map { addon -> launchLookup(addon.transportUrl) } + + var found: MetaDetail? = null + while (pending.isNotEmpty() && found == null) { + val (finished, result) = select>, AddonResourceResult>> { + pending.forEach { deferred -> deferred.onAwait { value -> deferred to value } } + } + pending.remove(finished) + when (result) { + is AddonResourceResult.Success -> found = result.value + is AddonResourceResult.Empty, + is AddonResourceResult.AddonUnsupported -> Unit + is AddonResourceResult.NetworkError -> Log.w( + "AddonRepository", + "Meta request failed: ${result.url} status=${result.statusCode}", + result.cause + ) + is AddonResourceResult.ParseError -> Log.w( + "AddonRepository", + "Meta parse failed: ${result.url}", + result.cause ) } - }.awaitAll() - } - - results.forEach { result -> - when (result) { - is AddonResourceResult.Success -> return result.value - is AddonResourceResult.Empty, - is AddonResourceResult.AddonUnsupported -> Unit - is AddonResourceResult.NetworkError -> Log.w( - "AddonRepository", - "Meta request failed: ${result.url} status=${result.statusCode}", - result.cause - ) - is AddonResourceResult.ParseError -> Log.w( - "AddonRepository", - "Meta parse failed: ${result.url}", - result.cause - ) } + pending.forEach { it.cancel() } + found } - return null } suspend fun getMetaDetailFromSpecificAddon(