fix(streams): proper usage of getBlocking

Author: slisson

datastructures/src/commonMain/kotlin/org/modelix/datastructures/btree/BTree.kt

@@ -1,8 +1,8 @@
 package org.modelix.datastructures.btree
 
 import org.modelix.datastructures.objects.IObjectGraph
-import org.modelix.kotlin.utils.DelicateModelixApi
 import org.modelix.streams.IStream
+import org.modelix.streams.getBlocking
 
 data class BTree<K, V>(val root: BTreeNode<K, V>) {
     constructor(config: BTreeConfig<K, V>) : this(BTreeNodeLeaf(config, emptyList()))
@@ -12,11 +12,10 @@ data class BTree<K, V>(val root: BTreeNode<K, V>) {
     fun validate() {
         graph.getStreamExecutor().query {
             root.validate(true)
-            @OptIn(DelicateModelixApi::class)
-            check(root.getEntries().toList().getSynchronous().map { it.key }.toSet().size == root.getEntries().map { it.key }.count().getSynchronous()) {
+            check(root.getEntries().toList().getBlocking(graph).map { it.key }.toSet().size == root.getEntries().map { it.key }.count().getBlocking(graph)) {
                 "duplicate entries: $root"
             }
-            check(root.getEntries().map { it.key }.toList().getSynchronous().sortedWith(root.config.keyConfiguration) == root.getEntries().map { it.key }.toList().getSynchronous()) {
+            check(root.getEntries().map { it.key }.toList().getBlocking(graph).sortedWith(root.config.keyConfiguration) == root.getEntries().map { it.key }.toList().getBlocking(graph)) {
                 "not sorted: $this"
             }
             IStream.of(Unit)

model-api/src/commonMain/kotlin/org/modelix/model/test/RandomModelChangeGenerator.kt

@@ -6,6 +6,7 @@ import org.modelix.model.api.async.IAsyncNode
 import org.modelix.model.api.async.asAsyncNode
 import org.modelix.model.api.getAncestors
 import org.modelix.streams.IStream
+import org.modelix.streams.getBlocking
 import org.modelix.streams.plus
 import kotlin.random.Random
 
@@ -99,7 +100,7 @@ class RandomModelChangeGenerator(val rootNode: INode, private val rand: Random)
             .map { it.asRegularNode() }
             .filter(condition)
             .firstOrNull()
-            .getSynchronous()
+            .getBlocking(rootNode.asAsyncNode())
     }
 }
 

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

@@ -54,7 +54,7 @@ class ModelClientAsStore(client: IModelClientV2, val repositoryId: RepositoryId)
         ).exactlyOne()
     }
 
-    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Zero {
+    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Completable {
         return IStream.fromFlow<Nothing>(
             flow {
                 client.pushObjects(

model-datastructure/src/commonMain/kotlin/org/modelix/datastructures/model/ModelTreeBuilder.kt

@@ -15,6 +15,7 @@ import org.modelix.model.TreeId
 import org.modelix.model.api.INodeReference
 import org.modelix.model.api.ITree
 import org.modelix.model.api.PNodeReference
+import org.modelix.streams.getBlocking
 
 abstract class ModelTreeBuilder<NodeId> private constructor(protected val common: Common = Common()) {
     protected class Common {
@@ -35,13 +36,13 @@ abstract class ModelTreeBuilder<NodeId> private constructor(protected val common
 
             val config = HamtNode.Config(
                 graph = common.graph,
-                keyConfig = nodeIdType!!,
+                keyConfig = nodeIdType,
                 valueConfig = ObjectReferenceDataTypeConfiguration(common.graph, NodeObjectData.Deserializer(nodeIdType, common.treeId)),
             )
             return HamtInternalNode.createEmpty(config)
                 .put(root.data.id, root.ref, common.graph)
                 .orNull()
-                .getSynchronous()!!
+                .getBlocking(common.graph)!!
                 .let { HamtTree(it) }
                 .autoResolveValues()
                 .asModelTree(common.treeId)
@@ -61,7 +62,7 @@ abstract class ModelTreeBuilder<NodeId> private constructor(protected val common
                 keyConfig = nodeIdType,
                 valueConfig = ObjectReferenceDataTypeConfiguration(common.graph, NodeObjectData.Deserializer(nodeIdType, common.treeId)),
             )
-            return PatriciaTrie(config).put(root.data.id, root.ref).getSynchronous().autoResolveValues().asModelTree(common.treeId)
+            return PatriciaTrie(config).put(root.data.id, root.ref).getBlocking(common.graph).autoResolveValues().asModelTree(common.treeId)
         }
     }
 

model-datastructure/src/commonMain/kotlin/org/modelix/model/async/BulkAsyncStore.kt

@@ -58,7 +58,7 @@ class BulkAsyncStore(
         return getAllAsStream(IStream.many(keys)).toMap({ it.first }, { it.second })
     }
 
-    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Zero {
+    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Completable {
         return store.putAll(entries)
     }
 }

model-datastructure/src/commonMain/kotlin/org/modelix/model/async/CachingAsyncStore.kt

@@ -80,7 +80,7 @@ class CachingAsyncStore(val store: IAsyncObjectStore, cacheSize: Int = 100_000)
         }
     }
 
-    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Zero {
+    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Completable {
         runSynchronized(cache) {
             for (entry in entries) {
                 cache.set(entry.key, entry.value)

model-datastructure/src/commonMain/kotlin/org/modelix/model/async/IAsyncObjectStore.kt

@@ -18,7 +18,7 @@ interface IAsyncObjectStore : IStreamExecutorProvider {
 
     fun getAllAsStream(keys: IStream.Many<ObjectRequest<*>>): IStream.Many<Pair<ObjectRequest<*>, IObjectData?>>
     fun getAllAsMap(keys: List<ObjectRequest<*>>): IStream.One<Map<ObjectRequest<*>, IObjectData?>>
-    fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Zero
+    fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Completable
 
     fun clearCache()
 

model-datastructure/src/commonMain/kotlin/org/modelix/model/async/LegacyKeyValueStoreAsAsyncStore.kt

@@ -50,7 +50,7 @@ class LegacyKeyValueStoreAsAsyncStore(val store: IKeyValueStore) : IAsyncObjectS
         )
     }
 
-    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Zero {
+    override fun putAll(entries: Map<ObjectRequest<*>, IObjectData>): IStream.Completable {
         store.putAll(entries.entries.associate { it.key.hash to it.value.serialize() })
         return IStream.zero()
     }

model-datastructure/src/commonMain/kotlin/org/modelix/model/lazy/CLTree.kt

@@ -29,6 +29,7 @@ import org.modelix.model.persistent.CPTree
 import org.modelix.streams.IStream
 import org.modelix.streams.IStreamExecutor
 import org.modelix.streams.IStreamExecutorProvider
+import org.modelix.streams.getBlocking
 
 private fun createNewTreeData(
     graph: IObjectGraph,
@@ -53,7 +54,7 @@ private fun createNewTreeData(
             HamtInternalNode.createEmpty(config)
                 .put(root.id, graph.fromCreated(root), graph)
                 .orNull()
-                .getSynchronous()!!,
+                .getBlocking(graph)!!,
         ),
         trieWithNodeRefIds = null,
         usesRoleIds = useRoleIds,

model-datastructure/src/commonMain/kotlin/org/modelix/model/lazy/OperationsCompressor.kt

@@ -16,6 +16,7 @@ import org.modelix.model.operations.SetPropertyOp
 import org.modelix.model.operations.SetReferenceOp
 import org.modelix.model.operations.UndoOp
 import org.modelix.model.persistent.CPTree
+import org.modelix.streams.getBlocking
 
 class OperationsCompressor(val resultTree: Object<CPTree>) {
 
@@ -55,7 +56,7 @@ class OperationsCompressor(val resultTree: Object<CPTree>) {
         }
 
         for (id in createdNodes) {
-            if (!resultTree.data.getModelTree().containsNode(id).getSynchronous()) {
+            if (!resultTree.data.getModelTree().containsNode(id).getBlocking(resultTree.graph)) {
                 throw RuntimeException("Tree expected to contain node $id")
             }
         }