From 611ca97ef0b7d8b86b9012ae0148c3965ca245ef Mon Sep 17 00:00:00 2001
From: giacomoaccursi <giacomo.accursi@gmail.com>
Date: Fri, 12 Jan 2024 18:13:43 +0100
Subject: [PATCH] chore: create simple custom flow

---
 src/main/kotlin/flow/CustomFlow.kt | 62 ++++++++++++++++++++++++++++++
 1 file changed, 62 insertions(+)
 create mode 100644 src/main/kotlin/flow/CustomFlow.kt

diff --git a/src/main/kotlin/flow/CustomFlow.kt b/src/main/kotlin/flow/CustomFlow.kt
new file mode 100644
index 0000000..6363225
--- /dev/null
+++ b/src/main/kotlin/flow/CustomFlow.kt
@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2024. Accursi Giacomo
+ *
+ * Use of this source code is governed by an MIT-style
+ * license that can be found in the LICENSE file or at
+ * https://opensource.org/licenses/MIT.
+ */
+
+package flow
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.withContext
+import java.util.concurrent.CountDownLatch
+
+/**
+ * A custom mutableSharedFlow waiting for the consumption of an emitted element.
+ */
+class CustomMutableSharedFlow<T>(private val flow: MutableSharedFlow<T>) : MutableSharedFlow<T> by flow {
+
+    private lateinit var latch: CountDownLatch
+
+    override suspend fun emit(value: T) {
+        latch = CountDownLatch(subscriptionCount.value)
+        flow.emit(value)
+        withContext(Dispatchers.IO) {
+            latch.await()
+        }
+        latch = CountDownLatch(0)
+    }
+
+    /**
+     * A function for notifying the consumption.
+     */
+    fun notifyConsumed() {
+        latch.countDown()
+    }
+}
+
+/**
+ * A custom mutableStateFlow waiting for the consumption of an emitted element.
+ */
+class CustomMutableStateFlow<T>(private val flow: MutableStateFlow<T>) : MutableStateFlow<T> by flow {
+    private lateinit var latch: CountDownLatch
+
+    override suspend fun emit(value: T) {
+        latch = CountDownLatch(subscriptionCount.value)
+        flow.emit(value)
+        withContext(Dispatchers.IO) {
+            latch.await()
+        }
+        latch = CountDownLatch(0)
+    }
+
+    /**
+     * A function for notifying the consumption.
+     */
+    fun notifyConsumed() {
+        latch.countDown()
+    }
+}