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

slisson · slisson · commit 4eafd32383fe · 2025-09-13T19:51:54.000+02:00
The Reaktive based implementation had the semantics of flatMapUnordered while all other
implementations had the one of flatMapOrdered.
diff --git a/model-api/src/commonMain/kotlin/org/modelix/model/api/async/AsyncNode.kt b/model-api/src/commonMain/kotlin/org/modelix/model/api/async/AsyncNode.kt
@@ -85,10 +85,17 @@ class AsyncNode(
 
     override fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode> {
         return if (includeSelf) {
-            getStreamExecutor()
             IStream.of(IStream.of(this), getDescendants(false)).flatten()
         } else {
-            getAllChildren().flatMap { it.getDescendants(true) }
+            getAllChildren().flatMapOrdered { it.getDescendants(true) }
+        }
+    }
+
+    override fun getDescendantsUnordered(includeSelf: Boolean): IStream.Many<IAsyncNode> {
+        return if (includeSelf) {
+            IStream.of(IStream.of(this), getDescendantsUnordered(false)).flatten()
+        } else {
+            getAllChildren().flatMapUnordered { it.getDescendantsUnordered(true) }
         }
     }
 }
diff --git a/model-api/src/commonMain/kotlin/org/modelix/model/api/async/IAsyncNode.kt b/model-api/src/commonMain/kotlin/org/modelix/model/api/async/IAsyncNode.kt
@@ -33,6 +33,7 @@ interface IAsyncNode : IStreamExecutorProvider {
     fun getAllReferenceTargets(): IStream.Many<Pair<IReferenceLinkReference, IAsyncNode>>
 
     fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode>
+    fun getDescendantsUnordered(includeSelf: Boolean): IStream.Many<IAsyncNode> = getDescendants(includeSelf)
 }
 
 interface INodeWithAsyncSupport : INode {
diff --git a/model-server/src/test/kotlin/org/modelix/model/server/LazyLoadingTest.kt b/model-server/src/test/kotlin/org/modelix/model/server/LazyLoadingTest.kt
@@ -148,7 +148,7 @@ private object DepthFirstSearchPattern : AccessPattern {
 private object StreamBasedApi : AccessPattern {
     override suspend fun runPattern(rootNode: INode) {
         val asyncNode = rootNode.asAsyncNode()
-        asyncNode.querySuspending { asyncNode.getDescendants(true).count() }
+        asyncNode.querySuspending { asyncNode.getDescendantsUnordered(true).count() }
     }
 }
 
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
`@@ -85,10 +85,17 @@ class AsyncNode(`
`85`	`85`
`86`	`86`	`override fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode> {`
`87`	`87`	`return if (includeSelf) {`
`88`		`- getStreamExecutor()`
`89`	`88`	`IStream.of(IStream.of(this), getDescendants(false)).flatten()`
`90`	`89`	`} else {`
`91`		`- getAllChildren().flatMap { it.getDescendants(true) }`
	`90`	`+ getAllChildren().flatMapOrdered { it.getDescendants(true) }`
	`91`	`+ }`
	`92`	`+ }`
	`93`	`+`
	`94`	`+ override fun getDescendantsUnordered(includeSelf: Boolean): IStream.Many<IAsyncNode> {`
	`95`	`+ return if (includeSelf) {`
	`96`	`+ IStream.of(IStream.of(this), getDescendantsUnordered(false)).flatten()`
	`97`	`+ } else {`
	`98`	`+ getAllChildren().flatMapUnordered { it.getDescendantsUnordered(true) }`
`92`	`99`	`}`
`93`	`100`	`}`
`94`	`101`	`}`
Original file line number	Diff line number	Diff line change
`@@ -33,6 +33,7 @@ interface IAsyncNode : IStreamExecutorProvider {`
`33`	`33`	`fun getAllReferenceTargets(): IStream.Many<Pair<IReferenceLinkReference, IAsyncNode>>`
`34`	`34`
`35`	`35`	`fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode>`
	`36`	`+ fun getDescendantsUnordered(includeSelf: Boolean): IStream.Many<IAsyncNode> = getDescendants(includeSelf)`
`36`	`37`	`}`
`37`	`38`
`38`	`39`	`interface INodeWithAsyncSupport : INode {`
Original file line number	Diff line number	Diff line change
`@@ -148,7 +148,7 @@ private object DepthFirstSearchPattern : AccessPattern {`
`148`	`148`	`private object StreamBasedApi : AccessPattern {`
`149`	`149`	`override suspend fun runPattern(rootNode: INode) {`
`150`	`150`	`val asyncNode = rootNode.asAsyncNode()`
`151`		`- asyncNode.querySuspending { asyncNode.getDescendants(true).count() }`
	`151`	`+ asyncNode.querySuspending { asyncNode.getDescendantsUnordered(true).count() }`
`152`	`152`	`}`
`153`	`153`	`}`
`154`	`154`
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> {`