Experimental: FlowLiveData

app/src/main/java/org/koitharu/kotatsu/reader/ui/ReaderViewModel.kt

@@ -33,7 +33,7 @@ import org.koitharu.kotatsu.reader.domain.ChaptersLoader
 import org.koitharu.kotatsu.reader.domain.PageLoader
 import org.koitharu.kotatsu.reader.ui.pager.ReaderUiState
 import org.koitharu.kotatsu.utils.SingleLiveEvent
-import org.koitharu.kotatsu.utils.ext.asLiveDataDistinct
+import org.koitharu.kotatsu.utils.asFlowLiveData
 import org.koitharu.kotatsu.utils.ext.printStackTraceDebug
 import org.koitharu.kotatsu.utils.ext.processLifecycleScope
 import org.koitharu.kotatsu.utils.ext.requireValue
@@ -86,7 +86,7 @@ class ReaderViewModel @AssistedInject constructor(
 	) { manga, policy ->
 		policy == ScreenshotsPolicy.BLOCK_ALL ||
 			(policy == ScreenshotsPolicy.BLOCK_NSFW && manga != null && manga.isNsfw)
-	}.asLiveDataDistinct(viewModelScope.coroutineContext + Dispatchers.Default, false)
+	}.asFlowLiveData(viewModelScope.coroutineContext + Dispatchers.Default, false)
 
 	val onZoomChanged = SingleLiveEvent<Unit>()
 
@@ -98,7 +98,7 @@ class ReaderViewModel @AssistedInject constructor(
 			bookmarksRepository.observeBookmark(manga, state.chapterId, state.page)
 				.map { it != null }
 		}
-	}.asLiveDataDistinct(viewModelScope.coroutineContext + Dispatchers.Default, false)
+	}.asFlowLiveData(viewModelScope.coroutineContext + Dispatchers.Default, false)
 
 	init {
 		loadImpl()

app/src/main/java/org/koitharu/kotatsu/utils/FlowLiveData.kt

@@ -0,0 +1,75 @@
+package org.koitharu.kotatsu.utils
+
+import androidx.lifecycle.LiveData
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.EmptyCoroutineContext
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.StateFlow
+
+private const val DEFAULT_TIMEOUT = 5_000L
+
+/**
+ * Similar to a CoroutineLiveData but optimized for using within infinite flows
+ */
+class FlowLiveData<T>(
+	private val flow: Flow<T>,
+	defaultValue: T,
+	context: CoroutineContext = EmptyCoroutineContext,
+	private val timeoutInMs: Long = DEFAULT_TIMEOUT,
+) : LiveData<T>(defaultValue) {
+
+	private val scope = CoroutineScope(Dispatchers.Main.immediate + context + SupervisorJob(context[Job]))
+	private var job: Job? = null
+	private var cancellationJob: Job? = null
+
+	override fun onActive() {
+		super.onActive()
+		cancellationJob?.cancel()
+		cancellationJob = null
+		if (job?.isActive == true) {
+			return
+		}
+		job = scope.launch {
+			flow.collect(Collector())
+		}
+	}
+
+	override fun onInactive() {
+		super.onInactive()
+		cancellationJob?.cancel()
+		cancellationJob = scope.launch(Dispatchers.Main.immediate) {
+			delay(timeoutInMs)
+			if (!hasActiveObservers()) {
+				job?.cancel()
+				job = null
+			}
+		}
+	}
+
+	private inner class Collector : FlowCollector<T> {
+
+		private var previousValue: Any? = value
+
+		override suspend fun emit(value: T) {
+			if (previousValue != value) {
+				previousValue = value
+				withContext(Dispatchers.Main.immediate) {
+					setValue(value)
+				}
+			}
+		}
+	}
+}
+
+fun <T> Flow<T>.asFlowLiveData(
+	context: CoroutineContext = EmptyCoroutineContext,
+	defaultValue: T,
+	timeoutInMs: Long = DEFAULT_TIMEOUT,
+): LiveData<T> = FlowLiveData(this, defaultValue, context, timeoutInMs)
+
+fun <T> StateFlow<T>.asFlowLiveData(
+	context: CoroutineContext = EmptyCoroutineContext,
+	timeoutInMs: Long = DEFAULT_TIMEOUT,
+): LiveData<T> = FlowLiveData(this, value, context, timeoutInMs)