fix(stream): IStream.flatMap replaced by flatMapOrdered/flatMapUnordered

slisson · slisson · commit 2f95bcd4ea89 · 2025-09-13T14:05:53.000+02:00
The Reaktive based implementation had the semantics of flatMapUnordered while all other
implementations had the one of flatMapOrdered.
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/CollectionAsStream.kt b/streams/src/commonMain/kotlin/org/modelix/streams/CollectionAsStream.kt
@@ -27,8 +27,12 @@ open class CollectionAsStream<E>(val collection: Collection<E>) : IStream.Many<E
         return SingleValueStream(collection.firstOrNull())
     }
 
-    override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
-        return convertLater().flatMap(mapper)
+    override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        return convertLater().flatMapOrdered(mapper)
+    }
+
+    override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        return convertLater().flatMapUnordered(mapper)
     }
 
     override fun <R> flatMapIterable(mapper: (E) -> Iterable<R>): IStream.Many<R> {
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/DeferredStreamBuilder.kt b/streams/src/commonMain/kotlin/org/modelix/streams/DeferredStreamBuilder.kt
@@ -162,9 +162,15 @@ class DeferredStreamBuilder : IStreamBuilder {
             return ConvertibleMany { convert(it).map(mapper) }
         }
 
-        override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
             return ConvertibleMany { converter ->
-                convert(converter).flatMap { mapper(it).convert(converter) }
+                convert(converter).flatMapUnordered { mapper(it).convert(converter) }
+            }
+        }
+
+        override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+            return ConvertibleMany { converter ->
+                convert(converter).flatMapOrdered { mapper(it).convert(converter) }
             }
         }
 
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/EmptyStream.kt b/streams/src/commonMain/kotlin/org/modelix/streams/EmptyStream.kt
@@ -69,7 +69,7 @@ class EmptyStream<E> : IStreamInternal.ZeroOrOne<E> {
     @DelicateModelixApi
     override fun iterateBlocking(visitor: (E) -> Unit) {}
 
-    override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+    override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
         return EmptyStream()
     }
 
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/FlowStreamBuilder.kt b/streams/src/commonMain/kotlin/org/modelix/streams/FlowStreamBuilder.kt
@@ -13,6 +13,7 @@ import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.flatMapConcat
+import kotlinx.coroutines.flow.flatMapMerge
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.fold
@@ -187,7 +188,11 @@ class FlowStreamBuilder() : IStreamBuilder {
             return Wrapper(wrapped.flatMapConcat { convert(mapper(it)) })
         }
 
-        override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+            return Wrapper(wrapped.flatMapMerge { convert(mapper(it)) })
+        }
+
+        override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
             return Wrapper(wrapped.flatMapConcat { convert(mapper(it)) })
         }
 
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/IStream.kt b/streams/src/commonMain/kotlin/org/modelix/streams/IStream.kt
@@ -27,7 +27,22 @@ interface IStream<out E> {
         fun filter(predicate: (E) -> Boolean): Many<E>
         fun <R> map(mapper: (E) -> R): Many<R>
         fun <R : Any> mapNotNull(mapper: (E) -> R?): Many<R> = map(mapper).filterNotNull()
-        fun <R> flatMap(mapper: (E) -> Many<R>): Many<R>
+
+        @Deprecated("Use flatMapOrdered or flatMapUnordered")
+        fun <R> flatMap(mapper: (E) -> Many<R>): Many<R> = flatMapOrdered(mapper)
+
+        /**
+         * Output elements are only emitted after all output elements of the previous input are emitted.
+         * Can have a lower performance than [flatMapUnordered].
+         */
+        fun <R> flatMapOrdered(mapper: (E) -> Many<R>): Many<R>
+
+        /**
+         * Output elements are emitted as soon as possible.
+         * Can have a higher performance than [flatMapOrdered].
+         */
+        fun <R> flatMapUnordered(mapper: (E) -> Many<R>): Many<R> = flatMapOrdered(mapper)
+
         fun <R> flatMapIterable(mapper: (E) -> Iterable<R>): Many<R> = flatMap { IStream.many(mapper(it)) }
         fun concat(other: Many<@UnsafeVariance E>): Many<E>
         fun concat(other: OneOrMany<@UnsafeVariance E>): OneOrMany<E>
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/ReaktiveStreamBuilder.kt b/streams/src/commonMain/kotlin/org/modelix/streams/ReaktiveStreamBuilder.kt
@@ -261,8 +261,12 @@ class ReaktiveStreamBuilder() : IStreamBuilder {
             throw UnsupportedOperationException("Use IStreamExecutor.iterate")
         }
 
-        override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
-            return WrapperMany(wrapped.flatMap { mapper(it).toReaktive() })
+        override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+            return WrapperMany(wrapped.flatMap(maxConcurrency = Int.MAX_VALUE) { mapper(it).toReaktive() })
+        }
+
+        override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+            return WrapperMany(wrapped.flatMap(maxConcurrency = 1) { mapper(it).toReaktive() })
         }
 
         override fun concat(other: IStream.Many<E>): IStream.Many<E> {
@@ -356,7 +360,7 @@ class ReaktiveStreamBuilder() : IStreamBuilder {
         }
 
         override fun <R> flatMapOne(mapper: (E) -> IStream.One<R>): OneOrMany<R> {
-            return WrapperOneOrMany(wrapped.flatMapSingle { mapper(it).toReaktive() })
+            return WrapperOneOrMany(wrapped.flatMapSingle(maxConcurrency = 1) { mapper(it).toReaktive() })
         }
 
         override fun onErrorReturn(valueSupplier: (Throwable) -> E): OneOrMany<E> {
@@ -439,7 +443,7 @@ class ReaktiveStreamBuilder() : IStreamBuilder {
             return WrapperMaybe(wrapped.flatMapMaybe { mapper(it).toReaktive() })
         }
 
-        override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
             return WrapperMany(wrapped.flatMapObservable { mapper(it).toReaktive() })
         }
 
@@ -565,7 +569,7 @@ class ReaktiveStreamBuilder() : IStreamBuilder {
             throw UnsupportedOperationException("Use IStreamExecutor.iterate")
         }
 
-        override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
             return WrapperMany(wrapped.flatMapObservable { mapper(it).toReaktive() })
         }
 
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/SequenceAsStream.kt b/streams/src/commonMain/kotlin/org/modelix/streams/SequenceAsStream.kt
@@ -49,8 +49,8 @@ open class SequenceAsStream<E>(val wrapped: Sequence<E>) : IStream.Many<E>, IStr
         return SequenceAsStreamOneOrMany(wrapped.ifEmpty { sequenceOf(alternative()) })
     }
 
-    override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
-        return convertLater().flatMap(mapper)
+    override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        return convertLater().flatMapOrdered(mapper)
     }
 
     override fun concat(other: IStream.Many<E>): IStream.Many<E> {
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/SequenceStreamBuilder.kt b/streams/src/commonMain/kotlin/org/modelix/streams/SequenceStreamBuilder.kt
@@ -188,7 +188,7 @@ class SequenceStreamBuilder() : IStreamBuilder {
             return Wrapper(wrapped.flatMap { convert(mapper(it)) })
         }
 
-        override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+        override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
             return Wrapper(wrapped.flatMap { convert(mapper(it)) })
         }
 
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/SingleValueStream.kt b/streams/src/commonMain/kotlin/org/modelix/streams/SingleValueStream.kt
@@ -87,7 +87,7 @@ class SingleValueStream<E>(val value: E) : IStreamInternal.One<E> {
         return this
     }
 
-    override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
+    override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {
         return mapper(value)
     }
 

Original file line number	Diff line number	Diff line change
`@@ -27,8 +27,12 @@ open class CollectionAsStream<E>(val collection: Collection<E>) : IStream.Many<E`
`27`	`27`	`return SingleValueStream(collection.firstOrNull())`
`28`	`28`	`}`
`29`	`29`
`30`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`31`		`- return convertLater().flatMap(mapper)`
	`30`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`31`	`+ return convertLater().flatMapOrdered(mapper)`
	`32`	`+ }`
	`33`	`+`
	`34`	`+ override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`35`	`+ return convertLater().flatMapUnordered(mapper)`
`32`	`36`	`}`
`33`	`37`
`34`	`38`	`override fun <R> flatMapIterable(mapper: (E) -> Iterable<R>): IStream.Many<R> {`
Original file line number	Diff line number	Diff line change
`@@ -162,9 +162,15 @@ class DeferredStreamBuilder : IStreamBuilder {`
`162`	`162`	`return ConvertibleMany { convert(it).map(mapper) }`
`163`	`163`	`}`
`164`	`164`
`165`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`165`	`+ override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`166`	`166`	`return ConvertibleMany { converter ->`
`167`		`- convert(converter).flatMap { mapper(it).convert(converter) }`
	`167`	`+ convert(converter).flatMapUnordered { mapper(it).convert(converter) }`
	`168`	`+ }`
	`169`	`+ }`
	`170`	`+`
	`171`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`172`	`+ return ConvertibleMany { converter ->`
	`173`	`+ convert(converter).flatMapOrdered { mapper(it).convert(converter) }`
`168`	`174`	`}`
`169`	`175`	`}`
`170`	`176`
Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,7 @@ class EmptyStream<E> : IStreamInternal.ZeroOrOne<E> {`
`69`	`69`	`@DelicateModelixApi`
`70`	`70`	`override fun iterateBlocking(visitor: (E) -> Unit) {}`
`71`	`71`
`72`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`72`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`73`	`73`	`return EmptyStream()`
`74`	`74`	`}`
`75`	`75`
Original file line number	Diff line number	Diff line change
`@@ -261,8 +261,12 @@ class ReaktiveStreamBuilder() : IStreamBuilder {`
`261`	`261`	`throw UnsupportedOperationException("Use IStreamExecutor.iterate")`
`262`	`262`	`}`
`263`	`263`
`264`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`265`		`- return WrapperMany(wrapped.flatMap { mapper(it).toReaktive() })`
	`264`	`+ override fun <R> flatMapUnordered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`265`	`+ return WrapperMany(wrapped.flatMap(maxConcurrency = Int.MAX_VALUE) { mapper(it).toReaktive() })`
	`266`	`+ }`
	`267`	`+`
	`268`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`269`	`+ return WrapperMany(wrapped.flatMap(maxConcurrency = 1) { mapper(it).toReaktive() })`
`266`	`270`	`}`
`267`	`271`
`268`	`272`	`override fun concat(other: IStream.Many<E>): IStream.Many<E> {`
`@@ -356,7 +360,7 @@ class ReaktiveStreamBuilder() : IStreamBuilder {`
`356`	`360`	`}`
`357`	`361`
`358`	`362`	`override fun <R> flatMapOne(mapper: (E) -> IStream.One<R>): OneOrMany<R> {`
`359`		`- return WrapperOneOrMany(wrapped.flatMapSingle { mapper(it).toReaktive() })`
	`363`	`+ return WrapperOneOrMany(wrapped.flatMapSingle(maxConcurrency = 1) { mapper(it).toReaktive() })`
`360`	`364`	`}`
`361`	`365`
`362`	`366`	`override fun onErrorReturn(valueSupplier: (Throwable) -> E): OneOrMany<E> {`
`@@ -439,7 +443,7 @@ class ReaktiveStreamBuilder() : IStreamBuilder {`
`439`	`443`	`return WrapperMaybe(wrapped.flatMapMaybe { mapper(it).toReaktive() })`
`440`	`444`	`}`
`441`	`445`
`442`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`446`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`443`	`447`	`return WrapperMany(wrapped.flatMapObservable { mapper(it).toReaktive() })`
`444`	`448`	`}`
`445`	`449`
`@@ -565,7 +569,7 @@ class ReaktiveStreamBuilder() : IStreamBuilder {`
`565`	`569`	`throw UnsupportedOperationException("Use IStreamExecutor.iterate")`
`566`	`570`	`}`
`567`	`571`
`568`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`572`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`569`	`573`	`return WrapperMany(wrapped.flatMapObservable { mapper(it).toReaktive() })`
`570`	`574`	`}`
`571`	`575`
Original file line number	Diff line number	Diff line change
`@@ -49,8 +49,8 @@ open class SequenceAsStream<E>(val wrapped: Sequence<E>) : IStream.Many<E>, IStr`
`49`	`49`	`return SequenceAsStreamOneOrMany(wrapped.ifEmpty { sequenceOf(alternative()) })`
`50`	`50`	`}`
`51`	`51`
`52`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`53`		`- return convertLater().flatMap(mapper)`
	`52`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`53`	`+ return convertLater().flatMapOrdered(mapper)`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`override fun concat(other: IStream.Many<E>): IStream.Many<E> {`
Original file line number	Diff line number	Diff line change
`@@ -188,7 +188,7 @@ class SequenceStreamBuilder() : IStreamBuilder {`
`188`	`188`	`return Wrapper(wrapped.flatMap { convert(mapper(it)) })`
`189`	`189`	`}`
`190`	`190`
`191`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`191`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`192`	`192`	`return Wrapper(wrapped.flatMap { convert(mapper(it)) })`
`193`	`193`	`}`
`194`	`194`
Original file line number	Diff line number	Diff line change
`@@ -87,7 +87,7 @@ class SingleValueStream<E>(val value: E) : IStreamInternal.One<E> {`
`87`	`87`	`return this`
`88`	`88`	`}`
`89`	`89`
`90`		`- override fun <R> flatMap(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
	`90`	`+ override fun <R> flatMapOrdered(mapper: (E) -> IStream.Many<R>): IStream.Many<R> {`
`91`	`91`	`return mapper(value)`
`92`	`92`	`}`
`93`	`93`