diff --git a/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt b/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt index 60dfe6ca..9f315906 100644 --- a/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt +++ b/app/src/main/kotlin/com/subtracks/data/net/NetworkMode.kt @@ -52,3 +52,29 @@ fun networkMode(context: Context): Flow { awaitClose { if (registered) runCatching { connectivity?.unregisterNetworkCallback(callback) } } }.distinctUntilChanged() } + +fun networkAvailable(context: Context): Flow { + val connectivity = context.applicationContext.getSystemService(ConnectivityManager::class.java) + return callbackFlow { + fun push() { + val capabilities = + connectivity?.activeNetwork?.let { connectivity.getNetworkCapabilities(it) } + // INTERNET, not VALIDATED: a LAN-only server is reachable without internet validation. + trySend(capabilities?.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) == true) + } + val callback = + object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) = push() + + override fun onLost(network: Network) = push() + + override fun onCapabilitiesChanged( + network: Network, + capabilities: NetworkCapabilities, + ) = push() + } + val registered = runCatching { connectivity?.registerDefaultNetworkCallback(callback) }.isSuccess + push() + awaitClose { if (registered) runCatching { connectivity?.unregisterNetworkCallback(callback) } } + }.distinctUntilChanged() +} diff --git a/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt b/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt index aa06b67f..a03e8861 100644 --- a/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt +++ b/app/src/main/kotlin/com/subtracks/data/repo/PendingServerActionSink.kt @@ -10,6 +10,8 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex @@ -21,20 +23,23 @@ class PendingServerActionSink( private val activeSourceId: suspend () -> Long?, private val offline: StateFlow, offlinePreferences: Flow, + networkAvailable: Flow, scope: CoroutineScope, ) : ServerActionSink { private val lock = Mutex() init { scope.launch { - offlinePreferences.collect { isOffline -> - // The persisted value is authoritative, unlike the process flag's default at cold - // start, so wait for that flag to agree before replaying. - if (!isOffline) { - offline.first { !it } - flush() + combine(offlinePreferences, networkAvailable) { isOffline, online -> !isOffline && online } + .distinctUntilChanged() + .collect { ready -> + // The persisted flag and connectivity are authoritative, unlike the process + // flag's default at cold start, so wait for that flag to agree before replaying. + if (ready) { + offline.first { !it } + flush() + } } - } } } diff --git a/app/src/main/kotlin/com/subtracks/di/AppModule.kt b/app/src/main/kotlin/com/subtracks/di/AppModule.kt index 4f94f3f2..0e9fe92d 100644 --- a/app/src/main/kotlin/com/subtracks/di/AppModule.kt +++ b/app/src/main/kotlin/com/subtracks/di/AppModule.kt @@ -14,6 +14,7 @@ import com.subtracks.data.download.DownloadEngine import com.subtracks.data.download.DownloadNotifier import com.subtracks.data.download.OkHttpArtworkFetcher import com.subtracks.data.download.SystemDownloadEngine +import com.subtracks.data.net.networkAvailable import com.subtracks.data.net.networkMode import com.subtracks.data.prefs.UserPreferences import com.subtracks.data.prefs.createUserPreferences @@ -85,6 +86,7 @@ fun appModule( activeSourceId = get()::activeSourceIdOnce, offline = get().offline, offlinePreferences = get().offlineMode(), + networkAvailable = networkAvailable(context.applicationContext), scope = get(), ) } diff --git a/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt b/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt index 1608bd82..8d12b68c 100644 --- a/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt +++ b/app/src/test/kotlin/com/subtracks/data/repo/PendingServerActionSinkTest.kt @@ -15,9 +15,12 @@ import com.subtracks.data.source.StarType import com.subtracks.data.source.subsonic.SubsonicException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.test.UnconfinedTestDispatcher import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertThrows @@ -73,6 +76,7 @@ class PendingServerActionSinkTest { private lateinit var sink: PendingServerActionSink private val offline = MutableStateFlow(true) private val offlinePreferences = MutableStateFlow(true) + private val networkAvailable = MutableStateFlow(true) private var sourceId = 0L @Before @@ -93,6 +97,7 @@ class PendingServerActionSinkTest { activeSourceId = { sourceId }, offline = offline, offlinePreferences = offlinePreferences, + networkAvailable = networkAvailable, scope = scope, ) } @@ -131,6 +136,40 @@ class PendingServerActionSinkTest { assertEquals(0, db.pendingActionDao().pending(sourceId).size) } + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun aRestoredConnectionReplaysThroughTheCollector() = + runBlocking { + offline.value = false + networkAvailable.value = false + val reconnectScope = CoroutineScope(SupervisorJob() + UnconfinedTestDispatcher()) + try { + val reconnectable = + PendingServerActionSink( + delegate = delegate, + db = db, + activeSourceId = { sourceId }, + offline = offline, + offlinePreferences = flowOf(false), + networkAvailable = networkAvailable, + scope = reconnectScope, + ) + delegate.scrobbleFailure = IOException("down") + + reconnectable.scrobble("s1", 1_000) + assertEquals(1, db.pendingActionDao().pending(sourceId).size) + + delegate.scrobbleFailure = null + networkAvailable.value = true + + awaitUntil("the queued scrobble to replay") { delegate.calls.isNotEmpty() } + assertEquals(listOf(Call.Scrobble("s1", 1_000)), delegate.calls) + assertEquals(0, db.pendingActionDao().pending(sourceId).size) + } finally { + cancelAndJoinBlocking(reconnectScope) + } + } + @Test fun anOfflineUnstarCoalescesAwayAnEarlierStar() = runBlocking {