Add Flow.concatMapEager

Author: akarnokd

README.md

@@ -11,7 +11,7 @@ Extensions to the Kotlin Flow library.
 
 ```groovy
 dependencies {
-    implementation "com.github.akarnokd:kotlin-flow-extensions:0.0.9"
+    implementation "com.github.akarnokd:kotlin-flow-extensions:0.0.10"
 }
 ```
 
@@ -39,6 +39,7 @@ Table of contents
   - `Flow.takeUntil`
   - `Flow.onBackpressureDrop`
   - [`Flow.flatMapDrop`](#flowflatmapdrop)
+  - [`Flow.concatMapEager`](#flowconcatmapeager)
 - `ParallelFlow` operators (`FlowExtensions`)
   - `ParallelFlow.concatMap`
   - `ParallelFlow.filter`
@@ -280,4 +281,31 @@ concatArrayEager(
         range(6, 5)
 )
 .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+```
+
+## Flow.concatMapEager
+
+Maps the upstream values into [Flow]s and launches them all at once, then
+emits items from a source before items of the next are emitted.
+
+For example, given two inner sources, if the first is slow, the items of the second won't be emitted until the first has
+finished emitting its items. This operators allows all sources to generate items in parallel but then still emit those
+items in the order their respective `Flow`s are mapped in.
+
+Note that the upstream and each source is consumed in an unbounded manner and thus,
+depending on the speed of the current source and the collector, the operator may retain
+items longer and may use more memory during its execution.
+
+```kotlin
+range(1, 5)
+.concatMapEager {
+    range(it * 10, 5).onEach { delay(100) }
+}
+.assertResult(
+        10, 11, 12, 13, 14,
+        20, 21, 22, 23, 24,
+        30, 31, 32, 33, 34,
+        40, 41, 42, 43, 44,
+        50, 51, 52, 53, 54
+)
 ```

gradle.properties

@@ -1,5 +1,5 @@
 GROUP=com.github.akarnokd
-VERSION_NAME=0.0.9
+VERSION_NAME=0.0.10
 
 POM_ARTIFACT_ID=kotlin-flow-extensions
 POM_NAME=Kotlin Flow Extensions

src/main/kotlin/hu/akarnokd/kotlin/flow/FlowExtensions.kt

@@ -220,6 +220,17 @@ fun <T> mergeArray(vararg sources: Flow<T>) : Flow<T> = FlowMergeArray(sources)
 @FlowPreview
 fun <T> concatArrayEager(vararg sources: Flow<T>) : Flow<T> = FlowConcatArrayEager(sources)
 
+/**
+ * Maps the upstream values into [Flow]s and launches them all at once, then
+ * emits items from a source before items of the next are emitted.
+ * Note that the upstream and each source is consumed in an unbounded manner and thus,
+ * depending on the speed of the current source and the collector, the operator may retain
+ * items longer and may use more memory during its execution.
+ * @param mapper the suspendable function to turn an upstream item into a [Flow]
+ */
+@FlowPreview
+fun <T, R> Flow<T>.concatMapEager(mapper: suspend (T) -> Flow<R>) : Flow<R> = FlowConcatMapEager(this, mapper)
+
 // -----------------------------------------------------------------------------------------
 // Parallel Extensions
 // -----------------------------------------------------------------------------------------

src/main/kotlin/hu/akarnokd/kotlin/flow/impl/FlowConcatMapEager.kt

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2019-2020 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package hu.akarnokd.kotlin.flow.impl
+
+import hu.akarnokd.kotlin.flow.Resumable
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.AbstractFlow
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicBoolean
+
+@FlowPreview
+class FlowConcatMapEager<T, R>(private val source: Flow<T>, private val mapper : suspend (t: T) -> Flow<R>) : AbstractFlow<R>() {
+    override suspend fun collectSafely(collector: FlowCollector<R>) {
+        coroutineScope {
+            val resumeOutput = Resumable()
+            val innerQueue = ConcurrentLinkedQueue<InnerQueue<R>>()
+            val innerDone = AtomicBoolean()
+
+            launch {
+                try {
+                    source.collect {
+                        val f = mapper(it);
+                        val iq = InnerQueue<R>()
+                        innerQueue.offer(iq);
+                        resumeOutput.resume()
+                        launch {
+                            try {
+                                f.collect {
+                                    iq.queue.offer(it)
+                                    resumeOutput.resume()
+                                }
+                            } finally {
+                                iq.done.set(true)
+                                resumeOutput.resume()
+                            }
+                        }
+                    }
+                } finally {
+                    innerDone.set(true)
+                    resumeOutput.resume()
+                }
+            }
+
+            var iq : InnerQueue<R>? = null
+
+            while (isActive) {
+
+                if (iq == null) {
+                    val id = innerDone.get()
+                    iq = innerQueue.poll()
+
+                    if (id && iq == null) {
+                        break
+                    }
+                }
+
+                if (iq != null) {
+                    val d = iq.done.get()
+                    val v = iq.queue.poll()
+
+                    if (d && v == null) {
+                        iq = null
+                        continue
+                    }
+
+                    if (v != null) {
+                        collector.emit(v)
+                        continue
+                    }
+                }
+                resumeOutput.await()
+            }
+        }
+    }
+
+    class InnerQueue<R> {
+        val queue = ConcurrentLinkedQueue<R>()
+        val done = AtomicBoolean()
+    }
+}

src/test/kotlin/hu/akarnokd/kotlin/flow/impl/FlowConcatArrayEagerTest.kt

@@ -51,4 +51,20 @@ class FlowConcatArrayEagerTest {
                 .take(6)
                 .assertResult(1, 2, 3, 4, 5, 6)
     }
+
+    @Test
+    fun cancel() = runBlocking() {
+        var counter = AtomicInteger()
+        concatArrayEager(
+                range(1, 5).onEach {
+                    delay(200)
+                    counter.getAndIncrement()
+                }
+        )
+        .take(3)
+        .assertResult(1, 2, 3)
+
+        delay(1200)
+        assertEquals(3, counter.get())
+    }
 }

src/test/kotlin/hu/akarnokd/kotlin/flow/impl/FlowConcatMapEagerTest.kt

@@ -0,0 +1,66 @@
+package hu.akarnokd.kotlin.flow.impl
+
+import hu.akarnokd.kotlin.flow.*
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.runBlocking
+import org.junit.Test
+import java.lang.IllegalArgumentException
+import java.lang.RuntimeException
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.assertEquals
+
+@FlowPreview
+class FlowConcatMapEagerTest {
+    @Test
+    fun basic() = runBlocking {
+        range(1, 5)
+        .concatMapEager {
+            range(it * 10, 5).onEach { delay(100) }
+        }
+        .assertResult(
+                10, 11, 12, 13, 14,
+                20, 21, 22, 23, 24,
+                30, 31, 32, 33, 34,
+                40, 41, 42, 43, 44,
+                50, 51, 52, 53, 54
+        )
+    }
+
+    @Test
+    fun take() = runBlocking {
+        range(1, 5)
+                .concatMapEager {
+                    range(it * 10, 5).onEach { delay(100) }
+                }
+                .take(7)
+                .assertResult(
+                        10, 11, 12, 13, 14,
+                        20, 21
+                )
+    }
+
+    @Test
+    fun crashMapper() = runBlocking {
+        range(1, 5)
+        .concatMapEager<Int, Int> {
+            throw IllegalArgumentException()
+        }
+        .assertFailure<Int, IllegalArgumentException>(IllegalArgumentException::class.java)
+    }
+
+    @Test
+    fun crashInner() = runBlocking {
+        range(1, 1)
+                .concatMapEager<Int, Int> {
+                    flow {
+                        throw IllegalArgumentException()
+                    }
+                }
+                .assertFailure<Int, IllegalArgumentException>(IllegalArgumentException::class.java)
+    }
+}