refactor: replace Reaktive with own stream API

Gives us the flexibility to use sequences, flows or Reaktive depending on the use case.

For example, consuming a Reaktive as a flow in JS doesn't work well because of the missing
backpressure. If there is no lazy loading then sequences provide a better performance.

BREAKING CHANGE: The asynchronous stream based model API switch from Reaktive to a custom one.

Author: slisson

model-api/src/commonMain/kotlin/org/modelix/model/api/async/AsyncNode.kt

@@ -1,16 +1,5 @@
 package org.modelix.model.api.async
 
-import com.badoo.reaktive.maybe.Maybe
-import com.badoo.reaktive.maybe.map
-import com.badoo.reaktive.maybe.mapNotNull
-import com.badoo.reaktive.maybe.notNull
-import com.badoo.reaktive.observable.Observable
-import com.badoo.reaktive.observable.flatMap
-import com.badoo.reaktive.observable.map
-import com.badoo.reaktive.observable.notNull
-import com.badoo.reaktive.observable.observableOf
-import com.badoo.reaktive.single.Single
-import com.badoo.reaktive.single.map
 import org.modelix.model.api.ConceptReference
 import org.modelix.model.api.IChildLinkReference
 import org.modelix.model.api.IConcept
@@ -21,6 +10,8 @@ import org.modelix.model.api.IPropertyReference
 import org.modelix.model.api.IReferenceLinkReference
 import org.modelix.model.api.resolve
 import org.modelix.model.api.resolveInCurrentContext
+import org.modelix.streams.IStream
+import org.modelix.streams.flatten
 
 class AsyncNode(
     private val regularNode: INode,
@@ -33,63 +24,63 @@ class AsyncNode(
 
     private fun Long.asNode(): IAsyncNode = createNodeAdapter(this)
 
-    override fun getParent(): Maybe<IAsyncNode> {
+    override fun getParent(): IStream.ZeroOrOne<IAsyncNode> {
         return tree().getParent(nodeId).map { it.asNode() }
     }
 
-    override fun getConcept(): Single<IConcept> {
+    override fun getConcept(): IStream.One<IConcept> {
         return tree().getConceptReference(nodeId).map { it.resolve() }
     }
 
-    override fun getConceptRef(): Single<ConceptReference> {
+    override fun getConceptRef(): IStream.One<ConceptReference> {
         return tree().getConceptReference(nodeId)
     }
 
-    override fun getRoleInParent(): Single<IChildLinkReference> {
+    override fun getRoleInParent(): IStream.One<IChildLinkReference> {
         return tree().getRole(nodeId)
     }
 
-    override fun getPropertyValue(role: IPropertyReference): Maybe<String> {
+    override fun getPropertyValue(role: IPropertyReference): IStream.ZeroOrOne<String> {
         return tree().getPropertyValue(nodeId, role)
     }
 
-    override fun getAllPropertyValues(): Observable<Pair<IPropertyReference, String>> {
+    override fun getAllPropertyValues(): IStream.Many<Pair<IPropertyReference, String>> {
         return tree().getAllPropertyValues(nodeId)
     }
 
-    override fun getAllChildren(): Observable<IAsyncNode> {
+    override fun getAllChildren(): IStream.Many<IAsyncNode> {
         return tree().getAllChildren(nodeId).map { it.asNode() }
     }
 
-    override fun getChildren(role: IChildLinkReference): Observable<IAsyncNode> {
+    override fun getChildren(role: IChildLinkReference): IStream.Many<IAsyncNode> {
         return tree().getChildren(nodeId, role).map { it.asNode() }
     }
 
-    override fun getReferenceTarget(role: IReferenceLinkReference): Maybe<IAsyncNode> {
+    override fun getReferenceTarget(role: IReferenceLinkReference): IStream.ZeroOrOne<IAsyncNode> {
         return getReferenceTargetRef(role).mapNotNull {
             INodeResolutionScope.runWithAdditionalScope(regularNode.getArea()) {
                 it.resolveInCurrentContext()?.asAsyncNode()
             }
         }
     }
 
-    override fun getReferenceTargetRef(role: IReferenceLinkReference): Maybe<INodeReference> {
-        return tree().getReferenceTarget(nodeId, role).notNull()
+    override fun getReferenceTargetRef(role: IReferenceLinkReference): IStream.ZeroOrOne<INodeReference> {
+        return tree().getReferenceTarget(nodeId, role)
     }
 
-    override fun getAllReferenceTargetRefs(): Observable<Pair<IReferenceLinkReference, INodeReference>> {
+    override fun getAllReferenceTargetRefs(): IStream.Many<Pair<IReferenceLinkReference, INodeReference>> {
         return tree().getAllReferenceTargetRefs(nodeId)
     }
 
-    override fun getAllReferenceTargets(): Observable<Pair<IReferenceLinkReference, IAsyncNode>> {
-        return tree().getAllReferenceTargetRefs(nodeId).map {
-            it.first to (it.second.resolveInCurrentContext() ?: return@map null).asAsyncNode()
-        }.notNull()
+    override fun getAllReferenceTargets(): IStream.Many<Pair<IReferenceLinkReference, IAsyncNode>> {
+        return tree().getAllReferenceTargetRefs(nodeId).mapNotNull {
+            it.first to (it.second.resolveInCurrentContext() ?: return@mapNotNull null).asAsyncNode()
+        }
     }
 
-    override fun getDescendants(includeSelf: Boolean): Observable<IAsyncNode> {
+    override fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode> {
         return if (includeSelf) {
-            observableOf(observableOf(this), getDescendants(false)).flatMap { it }
+            IStream.of(IStream.of(this), getDescendants(false)).flatten()
         } else {
             getAllChildren().flatMap { it.getDescendants(true) }
         }

model-api/src/commonMain/kotlin/org/modelix/model/api/async/IAsyncNode.kt

@@ -1,37 +1,35 @@
 package org.modelix.model.api.async
 
-import com.badoo.reaktive.maybe.Maybe
-import com.badoo.reaktive.observable.Observable
-import com.badoo.reaktive.single.Single
 import org.modelix.model.api.ConceptReference
 import org.modelix.model.api.IChildLinkReference
 import org.modelix.model.api.IConcept
 import org.modelix.model.api.INode
 import org.modelix.model.api.INodeReference
 import org.modelix.model.api.IPropertyReference
 import org.modelix.model.api.IReferenceLinkReference
+import org.modelix.streams.IStream
 
 interface IAsyncNode {
     fun asRegularNode(): INode
 
-    fun getConcept(): Single<IConcept>
-    fun getConceptRef(): Single<ConceptReference>
+    fun getConcept(): IStream.One<IConcept>
+    fun getConceptRef(): IStream.One<ConceptReference>
 
-    fun getRoleInParent(): Single<IChildLinkReference>
-    fun getParent(): Maybe<IAsyncNode>
+    fun getRoleInParent(): IStream.One<IChildLinkReference>
+    fun getParent(): IStream.ZeroOrOne<IAsyncNode>
 
-    fun getPropertyValue(role: IPropertyReference): Maybe<String>
-    fun getAllPropertyValues(): Observable<Pair<IPropertyReference, String>>
+    fun getPropertyValue(role: IPropertyReference): IStream.ZeroOrOne<String>
+    fun getAllPropertyValues(): IStream.Many<Pair<IPropertyReference, String>>
 
-    fun getAllChildren(): Observable<IAsyncNode>
-    fun getChildren(role: IChildLinkReference): Observable<IAsyncNode>
+    fun getAllChildren(): IStream.Many<IAsyncNode>
+    fun getChildren(role: IChildLinkReference): IStream.Many<IAsyncNode>
 
-    fun getReferenceTarget(role: IReferenceLinkReference): Maybe<IAsyncNode>
-    fun getReferenceTargetRef(role: IReferenceLinkReference): Maybe<INodeReference>
-    fun getAllReferenceTargetRefs(): Observable<Pair<IReferenceLinkReference, INodeReference>>
-    fun getAllReferenceTargets(): Observable<Pair<IReferenceLinkReference, IAsyncNode>>
+    fun getReferenceTarget(role: IReferenceLinkReference): IStream.ZeroOrOne<IAsyncNode>
+    fun getReferenceTargetRef(role: IReferenceLinkReference): IStream.ZeroOrOne<INodeReference>
+    fun getAllReferenceTargetRefs(): IStream.Many<Pair<IReferenceLinkReference, INodeReference>>
+    fun getAllReferenceTargets(): IStream.Many<Pair<IReferenceLinkReference, IAsyncNode>>
 
-    fun getDescendants(includeSelf: Boolean): Observable<IAsyncNode>
+    fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode>
 }
 
 interface INodeWithAsyncSupport : INode {

model-api/src/commonMain/kotlin/org/modelix/model/api/async/IAsyncTree.kt

@@ -1,73 +1,68 @@
 package org.modelix.model.api.async
 
-import com.badoo.reaktive.maybe.Maybe
-import com.badoo.reaktive.maybe.flatMapObservable
-import com.badoo.reaktive.observable.Observable
-import com.badoo.reaktive.observable.concatWith
-import com.badoo.reaktive.observable.flatMap
-import com.badoo.reaktive.observable.observableOf
-import com.badoo.reaktive.single.Single
 import org.modelix.model.api.ConceptReference
 import org.modelix.model.api.IChildLinkReference
 import org.modelix.model.api.INodeReference
 import org.modelix.model.api.IPropertyReference
 import org.modelix.model.api.IReferenceLinkReference
 import org.modelix.model.api.ITree
+import org.modelix.streams.IStream
+import org.modelix.streams.plus
 
 interface IAsyncTree {
     fun asSynchronousTree(): ITree
-    fun getChanges(oldVersion: IAsyncTree, changesOnly: Boolean): Observable<TreeChangeEvent>
+    fun getChanges(oldVersion: IAsyncTree, changesOnly: Boolean): IStream.Many<TreeChangeEvent>
 
-    fun getConceptReference(nodeId: Long): Single<ConceptReference>
+    fun getConceptReference(nodeId: Long): IStream.One<ConceptReference>
 
-    fun containsNode(nodeId: Long): Single<Boolean>
+    fun containsNode(nodeId: Long): IStream.One<Boolean>
 
-    fun getParent(nodeId: Long): Maybe<Long>
-    fun getRole(nodeId: Long): Single<IChildLinkReference>
+    fun getParent(nodeId: Long): IStream.ZeroOrOne<Long>
+    fun getRole(nodeId: Long): IStream.One<IChildLinkReference>
 
-    fun getChildren(parentId: Long, role: IChildLinkReference): Observable<Long>
-    fun getChildRoles(sourceId: Long): Observable<IChildLinkReference>
-    fun getAllChildren(parentId: Long): Observable<Long>
+    fun getChildren(parentId: Long, role: IChildLinkReference): IStream.Many<Long>
+    fun getChildRoles(sourceId: Long): IStream.Many<IChildLinkReference>
+    fun getAllChildren(parentId: Long): IStream.Many<Long>
 
-    fun getPropertyValue(nodeId: Long, role: IPropertyReference): Maybe<String>
+    fun getPropertyValue(nodeId: Long, role: IPropertyReference): IStream.ZeroOrOne<String>
 
-    fun getPropertyRoles(sourceId: Long): Observable<IPropertyReference>
-    fun getAllPropertyValues(sourceId: Long): Observable<Pair<IPropertyReference, String>>
+    fun getPropertyRoles(sourceId: Long): IStream.Many<IPropertyReference>
+    fun getAllPropertyValues(sourceId: Long): IStream.Many<Pair<IPropertyReference, String>>
 
-    fun getAllReferenceTargetRefs(sourceId: Long): Observable<Pair<IReferenceLinkReference, INodeReference>>
-    fun getReferenceTarget(sourceId: Long, role: IReferenceLinkReference): Maybe<INodeReference>
-    fun getReferenceRoles(sourceId: Long): Observable<IReferenceLinkReference>
+    fun getAllReferenceTargetRefs(sourceId: Long): IStream.Many<Pair<IReferenceLinkReference, INodeReference>>
+    fun getReferenceTarget(sourceId: Long, role: IReferenceLinkReference): IStream.ZeroOrOne<INodeReference>
+    fun getReferenceRoles(sourceId: Long): IStream.Many<IReferenceLinkReference>
 }
 
 interface IAsyncMutableTree : IAsyncTree {
-    fun deleteNodes(nodeIds: LongArray): Single<IAsyncMutableTree>
-    fun moveChild(newParentId: Long, newRole: IChildLinkReference, newIndex: Int, childId: Long): Single<IAsyncMutableTree>
-    fun setConcept(nodeId: Long, concept: ConceptReference): Single<IAsyncMutableTree>
+    fun deleteNodes(nodeIds: LongArray): IStream.One<IAsyncMutableTree>
+    fun moveChild(newParentId: Long, newRole: IChildLinkReference, newIndex: Int, childId: Long): IStream.One<IAsyncMutableTree>
+    fun setConcept(nodeId: Long, concept: ConceptReference): IStream.One<IAsyncMutableTree>
 
-    fun setPropertyValue(nodeId: Long, role: IPropertyReference, value: String?): Single<IAsyncMutableTree>
+    fun setPropertyValue(nodeId: Long, role: IPropertyReference, value: String?): IStream.One<IAsyncMutableTree>
 
-    fun addNewChildren(parentId: Long, role: IChildLinkReference, index: Int, newIds: LongArray, concepts: Array<ConceptReference>): Single<IAsyncMutableTree>
+    fun addNewChildren(parentId: Long, role: IChildLinkReference, index: Int, newIds: LongArray, concepts: Array<ConceptReference>): IStream.One<IAsyncMutableTree>
 
-    fun setReferenceTarget(sourceId: Long, role: IReferenceLinkReference, target: INodeReference?): Single<IAsyncMutableTree>
-    fun setReferenceTarget(sourceId: Long, role: IReferenceLinkReference, targetId: Long): Single<IAsyncMutableTree>
+    fun setReferenceTarget(sourceId: Long, role: IReferenceLinkReference, target: INodeReference?): IStream.One<IAsyncMutableTree>
+    fun setReferenceTarget(sourceId: Long, role: IReferenceLinkReference, targetId: Long): IStream.One<IAsyncMutableTree>
 }
 
-fun IAsyncTree.getAncestors(nodeId: Long, includeSelf: Boolean): Observable<Long> {
+fun IAsyncTree.getAncestors(nodeId: Long, includeSelf: Boolean): IStream.Many<Long> {
     return if (includeSelf) {
-        observableOf(nodeId).concatWith(getAncestors(nodeId, false))
+        IStream.of(nodeId) + getAncestors(nodeId, false)
     } else {
-        getParent(nodeId).flatMapObservable { getAncestors(it, true) }
+        getParent(nodeId).flatMap { getAncestors(it, true) }
     }
 }
 
-fun IAsyncTree.getDescendants(nodeId: Long, includeSelf: Boolean): Observable<Long> {
+fun IAsyncTree.getDescendants(nodeId: Long, includeSelf: Boolean): IStream.Many<Long> {
     return if (includeSelf) getDescendantsAndSelf(nodeId) else getDescendants(nodeId)
 }
 
-fun IAsyncTree.getDescendants(nodeId: Long): Observable<Long> {
+fun IAsyncTree.getDescendants(nodeId: Long): IStream.Many<Long> {
     return getAllChildren(nodeId).flatMap { getDescendantsAndSelf(it) }
 }
 
-fun IAsyncTree.getDescendantsAndSelf(nodeId: Long): Observable<Long> {
-    return observableOf(nodeId).concatWith(getDescendants(nodeId))
+fun IAsyncTree.getDescendantsAndSelf(nodeId: Long): IStream.Many<Long> {
+    return IStream.of(nodeId) + getDescendants(nodeId)
 }

model-api/src/commonMain/kotlin/org/modelix/model/api/async/NodeAsAsyncNode.kt

@@ -3,11 +3,8 @@ package org.modelix.model.api.async
 import com.badoo.reaktive.maybe.Maybe
 import com.badoo.reaktive.maybe.maybeOf
 import com.badoo.reaktive.maybe.maybeOfEmpty
-import com.badoo.reaktive.observable.Observable
-import com.badoo.reaktive.observable.asObservable
 import com.badoo.reaktive.single.Single
 import com.badoo.reaktive.single.singleOf
-import com.badoo.reaktive.single.toSingle
 import org.modelix.model.api.ConceptReference
 import org.modelix.model.api.IChildLinkReference
 import org.modelix.model.api.IConcept
@@ -19,6 +16,7 @@ import org.modelix.model.api.asProperty
 import org.modelix.model.api.getDescendants
 import org.modelix.model.api.meta.NullConcept
 import org.modelix.model.api.toReference
+import org.modelix.streams.IStream
 
 class NodeAsAsyncNode(val node: INode) : IAsyncNode {
 
@@ -27,55 +25,55 @@ class NodeAsAsyncNode(val node: INode) : IAsyncNode {
 
     override fun asRegularNode(): INode = node
 
-    override fun getConcept(): Single<IConcept> {
-        return (node.concept ?: NullConcept).asMono()
+    override fun getConcept(): IStream.One<IConcept> {
+        return IStream.of((node.concept ?: NullConcept))
     }
 
-    override fun getConceptRef(): Single<ConceptReference> {
-        return ((node.getConceptReference() ?: NullConcept.getReference()) as ConceptReference).asMono()
+    override fun getConceptRef(): IStream.One<ConceptReference> {
+        return IStream.of(((node.getConceptReference() ?: NullConcept.getReference()) as ConceptReference))
     }
 
-    override fun getParent(): Maybe<IAsyncNode> {
-        return node.parent?.asAsyncNode().asOptionalMono()
+    override fun getParent(): IStream.ZeroOrOne<IAsyncNode> {
+        return IStream.ofNotNull(node.parent?.asAsyncNode())
     }
 
-    override fun getRoleInParent(): Single<IChildLinkReference> {
-        return node.getContainmentLink().toReference().toSingle()
+    override fun getRoleInParent(): IStream.One<IChildLinkReference> {
+        return IStream.of(node.getContainmentLink().toReference())
     }
 
-    override fun getPropertyValue(role: IPropertyReference): Maybe<String> {
-        return node.getPropertyValue(role.asProperty()).asOptionalMono()
+    override fun getPropertyValue(role: IPropertyReference): IStream.ZeroOrOne<String> {
+        return IStream.ofNotNull(node.getPropertyValue(role.asProperty()))
     }
 
-    override fun getAllChildren(): Observable<IAsyncNode> {
-        return node.allChildren.map { it.asAsyncNode() }.asObservable()
+    override fun getAllChildren(): IStream.Many<IAsyncNode> {
+        return IStream.many(node.allChildren.map { it.asAsyncNode() })
     }
 
-    override fun getChildren(role: IChildLinkReference): Observable<IAsyncNode> {
-        return node.getChildren(role.toLegacy()).map { it.asAsyncNode() }.asObservable()
+    override fun getChildren(role: IChildLinkReference): IStream.Many<IAsyncNode> {
+        return IStream.many(node.getChildren(role.toLegacy()).map { it.asAsyncNode() })
     }
 
-    override fun getReferenceTarget(role: IReferenceLinkReference): Maybe<IAsyncNode> {
-        return node.getReferenceTarget(role.toLegacy())?.asAsyncNode().asOptionalMono()
+    override fun getReferenceTarget(role: IReferenceLinkReference): IStream.ZeroOrOne<IAsyncNode> {
+        return IStream.ofNotNull(node.getReferenceTarget(role.toLegacy())?.asAsyncNode())
     }
 
-    override fun getReferenceTargetRef(role: IReferenceLinkReference): Maybe<INodeReference> {
-        return node.getReferenceTargetRef(role.toLegacy()).asOptionalMono()
+    override fun getReferenceTargetRef(role: IReferenceLinkReference): IStream.ZeroOrOne<INodeReference> {
+        return IStream.ofNotNull(node.getReferenceTargetRef(role.toLegacy()))
     }
 
-    override fun getAllReferenceTargetRefs(): Observable<Pair<IReferenceLinkReference, INodeReference>> {
-        return node.getAllReferenceTargetRefs().map { it.first.toReference() to it.second }.asObservable()
+    override fun getAllReferenceTargetRefs(): IStream.Many<Pair<IReferenceLinkReference, INodeReference>> {
+        return IStream.many(node.getAllReferenceTargetRefs().map { it.first.toReference() to it.second })
     }
 
-    override fun getAllPropertyValues(): Observable<Pair<IPropertyReference, String>> {
-        return node.getAllProperties().map { it.first.toReference() to it.second }.asObservable()
+    override fun getAllPropertyValues(): IStream.Many<Pair<IPropertyReference, String>> {
+        return IStream.many(node.getAllProperties().map { it.first.toReference() to it.second })
     }
 
-    override fun getAllReferenceTargets(): Observable<Pair<IReferenceLinkReference, IAsyncNode>> {
-        return node.getAllReferenceTargets().map { it.first.toReference() to it.second.asAsyncNode() }.asObservable()
+    override fun getAllReferenceTargets(): IStream.Many<Pair<IReferenceLinkReference, IAsyncNode>> {
+        return IStream.many(node.getAllReferenceTargets().map { it.first.toReference() to it.second.asAsyncNode() })
     }
 
-    override fun getDescendants(includeSelf: Boolean): Observable<IAsyncNode> {
-        return node.getDescendants(includeSelf).map { it.asAsyncNode() }.asIterable().asObservable()
+    override fun getDescendants(includeSelf: Boolean): IStream.Many<IAsyncNode> {
+        return IStream.many(node.getDescendants(includeSelf).map { it.asAsyncNode() })
     }
 }