Convert download cache/queue flows into SharedFlows

Fixes #8727

app/src/main/java/eu/kanade/tachiyomi/data/download/DownloadCache.kt

@@ -19,10 +19,12 @@ import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.flow.shareIn
 import kotlinx.coroutines.withTimeout
 import uy.kohesive.injekt.Injekt
 import uy.kohesive.injekt.api.get
@@ -47,7 +49,9 @@ class DownloadCache(
     private val scope = CoroutineScope(Dispatchers.IO)
 
     private val _changes: Channel<Unit> = Channel(Channel.UNLIMITED)
-    val changes = _changes.receiveAsFlow().onStart { emit(Unit) }
+    val changes = _changes.receiveAsFlow()
+        .onStart { emit(Unit) }
+        .shareIn(scope, SharingStarted.Eagerly, 1)
 
     private val notifier by lazy { DownloadNotifier(context) }
 

app/src/main/java/eu/kanade/tachiyomi/data/download/model/DownloadQueue.kt

@@ -10,9 +10,11 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.flow.shareIn
 import rx.Observable
 import rx.subjects.PublishSubject
 import java.util.concurrent.CopyOnWriteArrayList
@@ -27,7 +29,10 @@ class DownloadQueue(
     private val statusSubject = PublishSubject.create<Download>()
 
     private val _updates: Channel<Unit> = Channel(Channel.UNLIMITED)
-    val updates = _updates.receiveAsFlow().onStart { emit(Unit) }.map { queue }
+    val updates = _updates.receiveAsFlow()
+        .onStart { emit(Unit) }
+        .map { queue }
+        .shareIn(scope, SharingStarted.Eagerly, 1)
 
     fun addAll(downloads: List<Download>) {
         downloads.forEach { download ->