diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/AddonRuntimeAggregator.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/AddonRuntimeAggregator.kt index 2ae982e55..dcf718f46 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/AddonRuntimeAggregator.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/AddonRuntimeAggregator.kt @@ -11,25 +11,19 @@ class AddonRuntimeAggregator( stremioAddons: List, request: MovieRuntimeRequest ): List { - val streams = mutableListOf() - if (stremioAddons.isNotEmpty()) { - streams += addonRuntimes[RuntimeKind.STREMIO] - ?.resolveMovieStreams(stremioAddons, request) - .orEmpty() - } - return streams + if (stremioAddons.isEmpty()) return emptyList() + return addonRuntimes[RuntimeKind.STREMIO] + ?.resolveMovieStreams(stremioAddons, request) + .orEmpty() } suspend fun resolveEpisodeStreams( stremioAddons: List, request: EpisodeRuntimeRequest ): List { - val streams = mutableListOf() - if (stremioAddons.isNotEmpty()) { - streams += addonRuntimes[RuntimeKind.STREMIO] - ?.resolveEpisodeStreams(stremioAddons, request) - .orEmpty() - } - return streams + if (stremioAddons.isEmpty()) return emptyList() + return addonRuntimes[RuntimeKind.STREMIO] + ?.resolveEpisodeStreams(stremioAddons, request) + .orEmpty() } } diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt index 0a606fd42..fa87c29d2 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/CloudSyncCoordinator.kt @@ -8,6 +8,7 @@ import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.delay import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.launch +import java.util.concurrent.atomic.AtomicBoolean import javax.inject.Inject import javax.inject.Singleton @@ -18,42 +19,48 @@ class CloudSyncCoordinator @Inject constructor( private val authRepository: AuthRepository ) { private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + private val lifecycleLock = Any() private var collectorJob: Job? = null private var flushJob: Job? = null - @Volatile - private var started = false + private val started = AtomicBoolean(false) fun start() { - if (started) return - started = true - collectorJob = scope.launch { - invalidationBus.events.collectLatest { invalidation -> - if (authRepository.getCurrentUserId().isNullOrBlank()) return@collectLatest - cloudSyncRepository.markLocalStateDirty() - scheduleFlush(invalidation) + synchronized(lifecycleLock) { + if (!started.compareAndSet(false, true)) return + collectorJob = scope.launch { + invalidationBus.events.collectLatest { invalidation -> + if (authRepository.getCurrentUserId().isNullOrBlank()) return@collectLatest + cloudSyncRepository.markLocalStateDirty() + scheduleFlush(invalidation) + } } } } fun stop() { - started = false - collectorJob?.cancel() - flushJob?.cancel() - collectorJob = null - flushJob = null + synchronized(lifecycleLock) { + started.set(false) + collectorJob?.cancel() + flushJob?.cancel() + collectorJob = null + flushJob = null + } } private fun scheduleFlush(invalidation: CloudSyncInvalidation) { - flushJob?.cancel() - flushJob = scope.launch { - delay(debounceMsFor(invalidation.scope)) - if (authRepository.getCurrentUserId().isNullOrBlank()) return@launch - runCatching { cloudSyncRepository.pushToCloud() } - .onFailure { error -> - Log.w("CloudSyncCoordinator", "Cloud push failed after ${invalidation.scope}: ${error.message}") - cloudSyncRepository.markLocalStateDirty() - } + synchronized(lifecycleLock) { + if (!started.get()) return + flushJob?.cancel() + flushJob = scope.launch { + delay(debounceMsFor(invalidation.scope)) + if (authRepository.getCurrentUserId().isNullOrBlank()) return@launch + runCatching { cloudSyncRepository.pushToCloud() } + .onFailure { error -> + Log.w("CloudSyncCoordinator", "Cloud push failed after ${invalidation.scope}: ${error.message}") + cloudSyncRepository.markLocalStateDirty() + } + } } } diff --git a/app/src/main/kotlin/com/arflix/tv/data/repository/PlaybackTelemetryRepository.kt b/app/src/main/kotlin/com/arflix/tv/data/repository/PlaybackTelemetryRepository.kt index 312207e09..9f3bdc072 100644 --- a/app/src/main/kotlin/com/arflix/tv/data/repository/PlaybackTelemetryRepository.kt +++ b/app/src/main/kotlin/com/arflix/tv/data/repository/PlaybackTelemetryRepository.kt @@ -13,15 +13,17 @@ import javax.inject.Singleton class PlaybackTelemetryRepository @Inject constructor( @ApplicationContext private val context: Context ) { - private val startupSamplesKey = longPreferencesKey("telemetry_startup_samples_v1") - private val startupAvgMsKey = longPreferencesKey("telemetry_startup_avg_ms_v1") - private val startupRetriesKey = longPreferencesKey("telemetry_startup_retries_v1") - private val failoverAttemptsKey = longPreferencesKey("telemetry_failover_attempts_v1") - private val failoverSuccessesKey = longPreferencesKey("telemetry_failover_successes_v1") - private val longRebuffersKey = longPreferencesKey("telemetry_long_rebuffers_v1") - private val playbackFailuresKey = longPreferencesKey("telemetry_playback_failures_v1") - private val lastStartupMsKey = longPreferencesKey("telemetry_last_startup_ms_v1") - private val lastSessionRetriesKey = intPreferencesKey("telemetry_last_session_retries_v1") + private companion object { + val startupSamplesKey = longPreferencesKey("telemetry_startup_samples_v1") + val startupAvgMsKey = longPreferencesKey("telemetry_startup_avg_ms_v1") + val startupRetriesKey = longPreferencesKey("telemetry_startup_retries_v1") + val failoverAttemptsKey = longPreferencesKey("telemetry_failover_attempts_v1") + val failoverSuccessesKey = longPreferencesKey("telemetry_failover_successes_v1") + val longRebuffersKey = longPreferencesKey("telemetry_long_rebuffers_v1") + val playbackFailuresKey = longPreferencesKey("telemetry_playback_failures_v1") + val lastStartupMsKey = longPreferencesKey("telemetry_last_startup_ms_v1") + val lastSessionRetriesKey = intPreferencesKey("telemetry_last_session_retries_v1") + } suspend fun recordStartup(startupMs: Long, retries: Int, failoversBeforeStart: Int) { val safeStartup = startupMs.coerceAtLeast(0L)