Concurrent map for flows

kmath-coroutines/src/commonMain/kotlin/scientifik/kmath/CoroutinesExtra.kt

@@ -89,8 +89,8 @@ suspend fun <T> AsyncFlow<T>.collect(concurrency: Int, action: suspend (value: T
 
 @FlowPreview
 fun <T, R> Flow<T>.map(
-    concurrencyLevel: Int,
+    dispatcher: CoroutineDispatcher,
-    dispatcher: CoroutineDispatcher = Dispatchers.Default,
+    concurrencyLevel: Int = 16,
     bufferSize: Int = concurrencyLevel,
     transform: suspend (T) -> R
 ): Flow<R> {

kmath-coroutines/src/jvmTest/kotlin/scientifik/kmath/streaming/BufferFlowTest.kt

@@ -6,6 +6,7 @@ import org.junit.Test
 import scientifik.kmath.async
 import scientifik.kmath.collect
 import scientifik.kmath.map
+import java.util.concurrent.Executors
 
 
 @ExperimentalCoroutinesApi
@@ -13,30 +14,32 @@ import scientifik.kmath.map
 @FlowPreview
 class BufferFlowTest {
 
+    val dispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
+
     @Test(timeout = 2000)
-    fun concurrentMap() {
+    fun map() {
         runBlocking {
-            (1..20).asFlow().map(4) {
+            (1..20).asFlow().map( dispatcher) {
-                println("Started $it")
+                println("Started $it on ${Thread.currentThread().name}")
                 @Suppress("BlockingMethodInNonBlockingContext")
                 Thread.sleep(200)
                 it
             }.collect {
-                println("Completed $it")
+                println("Completed $it on ${Thread.currentThread().name}")
             }
         }
     }
 
     @Test(timeout = 2000)
-    fun mapParallel() {
+    fun async() {
         runBlocking {
-            (1..20).asFlow().async {
+            (1..20).asFlow().async(dispatcher) {
-                println("Started $it")
+                println("Started $it on ${Thread.currentThread().name}")
                 @Suppress("BlockingMethodInNonBlockingContext")
                 Thread.sleep(200)
                 it
             }.collect(4) {
-                println("Completed $it")
+                println("Completed $it on ${Thread.currentThread().name}")
             }
         }
     }