From 2b252e11b8b2bbe6e7c3c853b6fec7d94b0bacfc Mon Sep 17 00:00:00 2001
From: dahunsi <dahunsi@google.com>
Date: Thu, 28 Dec 2023 08:43:01 +0100
Subject: [PATCH] Use a channel for QueryFlowValve instead of a
 MutableSharedFlow

---
 .../com/tunjid/tiler/ConcurrentListTiler.kt       | 15 ++++++---------
 1 file changed, 6 insertions(+), 9 deletions(-)

diff --git a/library/tiler/src/commonMain/kotlin/com/tunjid/tiler/ConcurrentListTiler.kt b/library/tiler/src/commonMain/kotlin/com/tunjid/tiler/ConcurrentListTiler.kt
index 2a50635..0b0b53a 100644
--- a/library/tiler/src/commonMain/kotlin/com/tunjid/tiler/ConcurrentListTiler.kt
+++ b/library/tiler/src/commonMain/kotlin/com/tunjid/tiler/ConcurrentListTiler.kt
@@ -16,14 +16,14 @@
 
 package com.tunjid.tiler
 
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.AbstractFlow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
-import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.emptyFlow
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flattenMerge
 import kotlinx.coroutines.flow.flow
@@ -170,19 +170,16 @@ private class QueryFlowValve<Query, Item>(
 ) : suspend (Flow<Tile.Data<Query, Item>>?) -> Unit,
     Flow<Tile.Data<Query, Item>> by fetcher.toOutputDataFlow(query) {
 
-    private val mutableSharedFlow = MutableSharedFlow<Flow<Tile.Data<Query, Item>>?>()
+    private val channel = Channel<Flow<Tile.Data<Query, Item>>?>()
 
-    val outputFlow: Flow<Tile.Output<Query, Item>> = mutableSharedFlow
+    val outputFlow: Flow<Tile.Output<Query, Item>> = channel
+        .consumeAsFlow()
         .distinctUntilChanged()
         .takeWhile { it != null }
         .flatMapLatest { it ?: emptyFlow() }
 
     override suspend fun invoke(flow: Flow<Tile.Data<Query, Item>>?) {
-        // Suspend till the downstream is connected
-        if (mutableSharedFlow.subscriptionCount.value < 1) {
-            mutableSharedFlow.subscriptionCount.first { it > 0 }
-        }
-        mutableSharedFlow.emit(flow)
+        channel.send(flow)
     }
 }