feat(streams): introduction of IStreamExecutor

Author: slisson

commitlint.config.js

@@ -22,6 +22,7 @@ module.exports = {
         "mps-model-server",
         "mps-sync-plugin",
         "openapi",
+        "streams",
         "ts-model-api",
         "vue-model-api",
       ],

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

@@ -11,6 +11,7 @@ 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.IStreamExecutor
 import org.modelix.streams.flatten
 
 class AsyncNode(
@@ -20,6 +21,10 @@ class AsyncNode(
     private val createNodeAdapter: (Long) -> IAsyncNode,
 ) : IAsyncNode {
 
+    override fun getStreamExecutor(): IStreamExecutor {
+        return tree().getStreamExecutor()
+    }
+
     override fun asRegularNode(): INode = regularNode
 
     private fun Long.asNode(): IAsyncNode = createNodeAdapter(this)

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

@@ -8,9 +8,11 @@ import org.modelix.model.api.INodeReference
 import org.modelix.model.api.IPropertyReference
 import org.modelix.model.api.IReferenceLinkReference
 import org.modelix.streams.IStream
+import org.modelix.streams.IStreamExecutor
 
 interface IAsyncNode {
     fun asRegularNode(): INode
+    fun getStreamExecutor(): IStreamExecutor
 
     fun getConcept(): IStream.One<IConcept>
     fun getConceptRef(): IStream.One<ConceptReference>

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

@@ -7,10 +7,13 @@ 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.IStreamExecutor
 import org.modelix.streams.plus
 
 interface IAsyncTree {
     fun asSynchronousTree(): ITree
+    fun getStreamExecutor(): IStreamExecutor
+
     fun getChanges(oldVersion: IAsyncTree, changesOnly: Boolean): IStream.Many<TreeChangeEvent>
 
     fun getConceptReference(nodeId: Long): IStream.One<ConceptReference>

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

@@ -17,8 +17,14 @@ import org.modelix.model.api.getDescendants
 import org.modelix.model.api.meta.NullConcept
 import org.modelix.model.api.toReference
 import org.modelix.streams.IStream
+import org.modelix.streams.IStreamExecutor
+import org.modelix.streams.SimpleStreamExecutor
+import org.modelix.streams.withSequences
 
-class NodeAsAsyncNode(val node: INode) : IAsyncNode {
+open class NodeAsAsyncNode(val node: INode) : IAsyncNode {
+    override fun getStreamExecutor(): IStreamExecutor {
+        return SimpleStreamExecutor().withSequences()
+    }
 
     private fun <T : Any> T?.asOptionalMono(): Maybe<T> = if (this != null) maybeOf(this) else maybeOfEmpty()
     private fun <T> T.asMono(): Single<T> = singleOf(this)

model-client/src/commonMain/kotlin/org/modelix/model/client/GarbageFilteringStore.kt

@@ -5,9 +5,10 @@ import org.modelix.model.IKeyValueStore
 import org.modelix.model.IKeyValueStoreWrapper
 import org.modelix.model.api.runSynchronized
 import org.modelix.model.persistent.HashUtil
+import org.modelix.streams.IStreamExecutorProvider
 
 @Deprecated(message = "Replaced by NonWrittenEntry")
-class GarbageFilteringStore(private val store: IKeyValueStore) : IKeyValueStoreWrapper {
+class GarbageFilteringStore(private val store: IKeyValueStore) : IKeyValueStoreWrapper, IStreamExecutorProvider by store {
     private val pendingEntries: MutableMap<String?, String?> = HashMap()
 
     override fun get(key: String): String? {

model-client/src/commonMain/kotlin/org/modelix/model/client2/ModelClientV2.kt

@@ -73,7 +73,10 @@ import org.modelix.model.server.api.v2.VersionDeltaStreamV2
 import org.modelix.model.server.api.v2.asStream
 import org.modelix.modelql.client.ModelQLClient
 import org.modelix.modelql.core.IMonoStep
+import org.modelix.streams.IExecutableStream
 import org.modelix.streams.IStream
+import org.modelix.streams.SimpleStreamExecutor
+import org.modelix.streams.withSequences
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
 
@@ -298,35 +301,43 @@ class ModelClientV2(
         LOG.debug { "${clientId.toString(16)}.push($branch, $version, $baseVersion)" }
         require(version is CLVersion)
         require(baseVersion is CLVersion?)
-        return IStream.useSequencesSuspending {
-            version.write()
-            val objects = version.fullDiff(baseVersion)
-            // large HTTP requests and large Json objects don't scale well
-            val lastChunk = pushObjects(branch.repositoryId, objects.map { it.hash to it.serialize() }, returnLastChunk = true)
-            val delta = VersionDelta(version.getContentHash(), null, objectsMap = lastChunk.toMap())
-            httpClient.preparePost {
-                url {
-                    takeFrom(baseUrl)
-                    appendPathSegmentsEncodingSlash("repositories", branch.repositoryId.id, "branches", branch.branchName)
-                }
-                useVersionStreamFormat()
-                contentType(ContentType.Application.Json)
-                setBody(delta)
-            }.execute { response ->
-                createVersion(getStore(branch.repositoryId), version, response.readVersionDelta())
+        version.write()
+        val objects = version.asyncStore.getStreamExecutor().queryManyLater {
+            version.fullDiff(baseVersion).map { it.hash to it.serialize() }
+        }
+        // large HTTP requests and large Json objects don't scale well
+        val lastChunk = pushObjects(branch.repositoryId, objects, returnLastChunk = true)
+        val delta = VersionDelta(version.getContentHash(), null, objectsMap = lastChunk.toMap())
+        return httpClient.preparePost {
+            url {
+                takeFrom(baseUrl)
+                appendPathSegmentsEncodingSlash("repositories", branch.repositoryId.id, "branches", branch.branchName)
             }
+            useVersionStreamFormat()
+            contentType(ContentType.Application.Json)
+            setBody(delta)
+        }.execute { response ->
+            createVersion(getStore(branch.repositoryId), version, response.readVersionDelta())
         }
     }
 
     override suspend fun pushObjects(repository: RepositoryId, objects: Sequence<ObjectHashAndSerializedObject>) {
-        pushObjects(repository, IStream.many(objects), false)
+        pushObjects(
+            repository,
+            SimpleStreamExecutor().withSequences().queryManyLater { IStream.many(objects) },
+            false,
+        )
     }
 
     /**
      * If the last chunk is smaller than #minBodySize, then the remaining objects are returned for inlining in the
      * main request.
      */
-    private suspend fun pushObjects(repository: RepositoryId, objects: IStream.Many<ObjectHashAndSerializedObject>, returnLastChunk: Boolean): List<ObjectHashAndSerializedObject> {
+    private suspend fun pushObjects(
+        repository: RepositoryId,
+        objects: IExecutableStream.Many<ObjectHashAndSerializedObject>,
+        returnLastChunk: Boolean,
+    ): List<ObjectHashAndSerializedObject> {
         LOG.debug { "${clientId.toString(16)}.pushObjects($repository)" }
         val maxBodySize = 2 * 1024 * 1024
         val chunkContent = StringBuilder()
@@ -345,7 +356,7 @@ class ModelClientV2(
             chunkEntries.clear()
         }
 
-        objects.asSequence().forEach { entry ->
+        objects.iterateSuspending { entry ->
             val entrySize = (if (chunkContent.isEmpty()) 0 else 1) + entry.first.length + 1 + entry.second.length
             if (chunkContent.length + entrySize > maxBodySize) {
                 sendChunk()

model-client/src/commonTest/kotlin/org/modelix/model/api/TreePointerTest.kt

@@ -23,6 +23,6 @@ class TreePointerTest {
         val rootNode = branch.getRootNode()
         val role = IReferenceLinkReference.fromName("refA")
         rootNode.setReferenceTarget(role.toLegacy(), rootNode)
-        assertEquals(rootNode, rootNode.asAsyncNode().getReferenceTarget(role).getSynchronous()?.asRegularNode())
+        assertEquals(rootNode, rootNode.asAsyncNode().let { n -> n.getStreamExecutor().query { n.getReferenceTarget(role).orNull() } }?.asRegularNode())
     }
 }

model-client/src/jvmMain/kotlin/org/modelix/model/KeyValueStoreCache.kt

@@ -3,11 +3,12 @@ package org.modelix.model
 import org.apache.commons.collections4.map.LRUMap
 import org.modelix.model.persistent.HashUtil
 import org.modelix.model.util.StreamUtils.toStream
+import org.modelix.streams.IStreamExecutorProvider
 import java.util.Collections
 import java.util.concurrent.CompletableFuture
 import java.util.stream.Collectors
 
-class KeyValueStoreCache(private val store: IKeyValueStore) : IKeyValueStoreWrapper {
+class KeyValueStoreCache(private val store: IKeyValueStore) : IKeyValueStoreWrapper, IStreamExecutorProvider by store {
     private val cache = Collections.synchronizedMap(LRUMap<String, String?>(300000))
     private val pendingPrefetches: MutableSet<String> = HashSet()
     private val activeRequests: MutableList<GetRequest> = ArrayList()

model-client/src/jvmMain/kotlin/org/modelix/model/client/AsyncStore.kt

@@ -3,10 +3,11 @@ package org.modelix.model.client
 import org.modelix.model.IKeyListener
 import org.modelix.model.IKeyValueStore
 import org.modelix.model.IKeyValueStoreWrapper
+import org.modelix.streams.IStreamExecutorProvider
 import java.util.Objects
 import java.util.concurrent.atomic.AtomicBoolean
 
-class AsyncStore(private val store: IKeyValueStore) : IKeyValueStoreWrapper {
+class AsyncStore(private val store: IKeyValueStore) : IKeyValueStoreWrapper, IStreamExecutorProvider by store {
     private val consumerActive = AtomicBoolean()
     private val pendingWrites: MutableMap<String, String?> = LinkedHashMap()
     override fun get(key: String): String? {