Add storage API test (inclomplete)

build.gradle.kts

@@ -6,7 +6,7 @@ plugins {
     id("space.kscience.gradle.project")
 }
 
-val dataforgeVersion: String by extra("0.6.1")
+val dataforgeVersion: String by extra("0.6.2-dev-3")
 val ktorVersion: String by extra(space.kscience.gradle.KScienceVersions.ktorVersion)
 val rsocketVersion by extra("0.15.4")
 val xodusVersion by extra("2.0.1")

controls-core/src/commonMain/kotlin/space/kscience/controls/api/DeviceMessage.kt

@@ -1,7 +1,11 @@
+@file:OptIn(ExperimentalSerializationApi::class)
+
 package space.kscience.controls.api
 
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
+import kotlinx.serialization.EncodeDefault
+import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.Json
@@ -55,9 +59,9 @@ public data class PropertyChangedMessage(
     override val sourceDevice: Name = Name.EMPTY,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = block(sourceDevice))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = block(sourceDevice))
 }
 
 /**
@@ -71,9 +75,9 @@ public data class PropertySetMessage(
     override val sourceDevice: Name? = null,
     override val targetDevice: Name,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = sourceDevice?.let(block))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = sourceDevice?.let(block))
 }
 
 /**
@@ -87,9 +91,9 @@ public data class PropertyGetMessage(
     override val sourceDevice: Name? = null,
     override val targetDevice: Name,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = sourceDevice?.let(block))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = sourceDevice?.let(block))
 }
 
 /**
@@ -101,9 +105,9 @@ public data class GetDescriptionMessage(
     override val sourceDevice: Name? = null,
     override val targetDevice: Name,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = sourceDevice?.let(block))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = sourceDevice?.let(block))
 }
 
 /**
@@ -118,9 +122,9 @@ public data class DescriptionMessage(
     override val sourceDevice: Name,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = block(sourceDevice))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = block(sourceDevice))
 }
 
 /**
@@ -137,9 +141,9 @@ public data class ActionExecuteMessage(
     override val sourceDevice: Name? = null,
     override val targetDevice: Name,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = sourceDevice?.let(block))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = sourceDevice?.let(block))
 }
 
 /**
@@ -156,9 +160,9 @@ public data class ActionResultMessage(
     override val sourceDevice: Name,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = block(sourceDevice))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = block(sourceDevice))
 }
 
 /**
@@ -171,9 +175,9 @@ public data class BinaryNotificationMessage(
     override val sourceDevice: Name,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = block(sourceDevice))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = block(sourceDevice))
 }
 
 /**
@@ -186,9 +190,9 @@ public data class EmptyDeviceMessage(
     override val sourceDevice: Name? = null,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = sourceDevice?.let(block))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = sourceDevice?.let(block))
 }
 
 /**
@@ -202,9 +206,9 @@ public data class DeviceLogMessage(
     override val sourceDevice: Name? = null,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = sourceDevice?.let(block))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = sourceDevice?.let(block))
 }
 
 /**
@@ -219,9 +223,9 @@ public data class DeviceErrorMessage(
     override val sourceDevice: Name,
     override val targetDevice: Name? = null,
     override val comment: String? = null,
-    override val time: Instant? = Clock.System.now()
+    @EncodeDefault override val time: Instant? = Clock.System.now(),
-) : DeviceMessage(){
+) : DeviceMessage() {
-    override fun changeSource(block: (Name) -> Name):DeviceMessage  = copy(sourceDevice = block(sourceDevice))
+    override fun changeSource(block: (Name) -> Name): DeviceMessage = copy(sourceDevice = block(sourceDevice))
 }
 
 

demo/echo/src/main/kotlin/space/kscience/controls/demo/echo/main.kt

@@ -22,7 +22,7 @@ private suspend fun MagixEndpoint.collectEcho(scope: CoroutineScope, n: Int) {
     scope.launch {
         subscribe(
             MagixMessageFilter(
-                origin = listOf("loop")
+                source = listOf("loop")
             )
         ).collect { message ->
             if (message.id?.endsWith(".response") == true) {

demo/many-devices/build.gradle.kts

@@ -19,7 +19,7 @@ dependencies {
     implementation(projects.magix.magixZmq)
 
     implementation("io.ktor:ktor-client-cio:$ktorVersion")
-    implementation("space.kscience:plotlykt-server:0.5.3")
+    implementation("space.kscience:plotlykt-server:0.6.0")
     implementation(spclibs.logback.classic)
 }
 

demo/many-devices/src/main/kotlin/space/kscience/controls/demo/MassDevice.kt

@@ -1,11 +1,10 @@
 package space.kscience.controls.demo
 
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.*
-import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.isActive
+import kotlinx.coroutines.sync.Mutex
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.withLock
 import kotlinx.datetime.Clock
 import space.kscience.controls.client.connectToMagix
 import space.kscience.controls.client.magixFormat
@@ -20,6 +19,7 @@ import space.kscience.dataforge.meta.get
 import space.kscience.dataforge.meta.int
 import space.kscience.magix.api.MagixEndpoint
 import space.kscience.magix.api.subscribe
+import space.kscience.magix.rsocket.rSocketWithTcp
 import space.kscience.magix.rsocket.rSocketWithWebSockets
 import space.kscience.magix.server.RSocketMagixFlowPlugin
 import space.kscience.magix.server.startMagixServer
@@ -31,7 +31,6 @@ import space.kscience.plotly.server.PlotlyUpdateMode
 import space.kscience.plotly.server.serve
 import space.kscience.plotly.server.show
 import space.kscince.magix.zmq.ZmqMagixFlowPlugin
-import java.util.concurrent.ConcurrentHashMap
 import kotlin.random.Random
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
@@ -49,14 +48,15 @@ class MassDevice(context: Context, meta: Meta) : DeviceBySpec<MassDevice>(MassDe
         val value by doubleProperty { randomValue }
 
         override suspend fun MassDevice.onOpen() {
-            doRecurring(50.milliseconds) {
+            doRecurring(10.milliseconds) {
                 read(value)
             }
         }
     }
 }
 
-fun main() {
+@OptIn(DelicateCoroutinesApi::class)
+suspend fun main() {
     val context = Context("Mass")
 
     context.startMagixServer(
@@ -67,7 +67,7 @@ fun main() {
     val numDevices = 100
 
     repeat(numDevices) {
-        context.launch(Dispatchers.IO) {
+        context.launch(newFixedThreadPoolContext(2, "Device${it}")) {
             val deviceContext = Context("Device${it}") {
                 plugin(DeviceManager)
             }
@@ -77,7 +77,7 @@ fun main() {
             deviceManager.install("device$it", MassDevice)
 
             val endpointId = "device$it"
-            val deviceEndpoint = MagixEndpoint.rSocketWithWebSockets("localhost")
+            val deviceEndpoint = MagixEndpoint.rSocketWithTcp("localhost")
             deviceManager.connectToMagix(deviceEndpoint, endpointId)
         }
     }
@@ -94,14 +94,19 @@ fun main() {
                     launch(Dispatchers.IO) {
                         val monitorEndpoint = MagixEndpoint.rSocketWithWebSockets("localhost")
 
-                        val latest = ConcurrentHashMap<String, Duration>()
+                        val mutex = Mutex()
+
+                        val latest = HashMap<String, Duration>()
 
                         monitorEndpoint.subscribe(DeviceManager.magixFormat).onEach { (magixMessage, payload) ->
+                            mutex.withLock {
                                 latest[magixMessage.sourceEndpoint] = Clock.System.now() - payload.time!!
+                            }
                         }.launchIn(this)
 
                         while (isActive) {
                             delay(200)
+                            mutex.withLock {
                                 val sorted = latest.mapKeys { it.key.substring(6).toInt() }.toSortedMap()
                                 latest.clear()
                                 x.numbers = sorted.keys
@@ -112,6 +117,7 @@ fun main() {
                 }
             }
         }
+    }
 
     application.show()
 

magix/magix-api/src/commonMain/kotlin/space/kscience/magix/api/MagixFormat.kt

@@ -28,7 +28,7 @@ public fun <T> MagixEndpoint.subscribe(
     originFilter: Collection<String>? = null,
     targetFilter: Collection<String>? = null,
 ): Flow<Pair<MagixMessage, T>> = subscribe(
-    MagixMessageFilter(format = format.formats, origin = originFilter, target = targetFilter)
+    MagixMessageFilter(format = format.formats, source = originFilter, target = targetFilter)
 ).map {
     val value: T = magixJson.decodeFromJsonElement(format.serializer, it.payload)
     it to value

magix/magix-api/src/commonMain/kotlin/space/kscience/magix/api/MagixMessageFilter.kt

@@ -10,13 +10,13 @@ import kotlinx.serialization.Serializable
 @Serializable
 public data class MagixMessageFilter(
     val format: Collection<String>? = null,
-    val origin: Collection<String>? = null,
+    val source: Collection<String>? = null,
     val target: Collection<String>? = null,
 ) {
 
     public fun accepts(message: MagixMessage): Boolean =
         format?.contains(message.format) ?: true
-                && origin?.contains(message.sourceEndpoint) ?: true
+                && source?.contains(message.sourceEndpoint) ?: true
                 && target?.contains(message.targetEndpoint) ?: true
 
     public companion object {

magix/magix-rsocket/src/commonMain/kotlin/space/kscience/magix/rsocket/RSocketMagixEndpoint.kt

@@ -51,11 +51,12 @@ public class RSocketMagixEndpoint(private val rSocket: RSocket) : MagixEndpoint,
 }
 
 
-internal fun buildConnector(rSocketConfig: RSocketConnectorBuilder.ConnectionConfigBuilder.() -> Unit) =
+internal fun buildConnector(
-    RSocketConnector {
+    rSocketConfig: RSocketConnectorBuilder.ConnectionConfigBuilder.() -> Unit,
-        reconnectable(10)
+) = RSocketConnector {
+    reconnectable(5)
     connectionConfig(rSocketConfig)
-    }
+}
 
 /**
  * Build a websocket based endpoint connected to [host], [port] and given routing [path]

magix/magix-server/src/main/kotlin/space/kscience/magix/server/RSocketMagixFlowPlugin.kt

@@ -35,8 +35,11 @@ public class RSocketMagixFlowPlugin(
         receive: Flow<MagixMessage>,
         sendMessage: suspend (MagixMessage) -> Unit,
     ): Job {
-        val tcpTransport =
+        val tcpTransport = TcpServerTransport(
-            TcpServerTransport(hostname = serverHost, port = serverPort, configure = transportConfiguration)
+            hostname = serverHost,
+            port = serverPort,
+            configure = transportConfiguration
+        )
         val rSocketJob: TcpServer = RSocketServer(rsocketConfiguration)
             .bindIn(scope, tcpTransport, acceptor(scope, receive, sendMessage))
 
@@ -60,7 +63,7 @@ public class RSocketMagixFlowPlugin(
                         MagixMessageFilter.serializer(),
                         request.data.readText()
                     )
-                    request.close()
+
                     receive.filter(filter).map { message ->
                         val string = MagixEndpoint.magixJson.encodeToString(MagixMessage.serializer(), message)
                         buildPayload { data(string) }
@@ -72,7 +75,7 @@ public class RSocketMagixFlowPlugin(
                         MagixMessage.serializer(),
                         request.data.readText()
                     )
-                    request.close()
+
                     sendMessage(message)
                 }
                 // bidirectional connection, used for streaming connection
@@ -81,12 +84,12 @@ public class RSocketMagixFlowPlugin(
                         sendMessage(
                             MagixEndpoint.magixJson.decodeFromString(
                                 MagixMessage.serializer(),
-                                inputPayload.use{ it.data.readText()}
+                                inputPayload.use { it.data.readText() }
                             )
                         )
                     }.launchIn(this)
 
-                    val filterText = request.use { it.data.readText()}
+                    val filterText = request.use { it.data.readText() }
 
                     val filter = if (filterText.isNotBlank()) {
                         MagixEndpoint.magixJson.decodeFromString(MagixMessageFilter.serializer(), filterText)

magix/magix-storage/magix-storage-xodus/src/main/kotlin/space/kscience/magix/storage/xodus/XodusMagixStorage.kt

@@ -19,20 +19,25 @@ import space.kscience.magix.storage.test
 import java.nio.file.Path
 import kotlin.sequences.Sequence
 
-/**
- * Attach a Xodus storage process to the given endpoint.
- */
-public class XodusMagixStorage(
-    scope: CoroutineScope,
-    private val store: PersistentEntityStore,
-    endpoint: MagixEndpoint,
-    filter: MagixMessageFilter = MagixMessageFilter.ALL,
-) : MagixHistory, AutoCloseable {
 
-    //TODO consider message buffering
+private fun Entity.parseMagixMessage(): MagixMessage = MagixMessage(
-    internal val subscriptionJob = endpoint.subscribe(filter).onEach { message ->
+    format = getProperty(MagixMessage::format.name).toString(),
-        store.executeInTransaction { transaction ->
+    payload = getBlobString(MagixMessage::payload.name)?.let {
-            transaction.newEntity(MAGIC_MESSAGE_ENTITY_TYPE).apply {
+        magixJson.parseToJsonElement(it)
+    } ?: JsonObject(emptyMap()),
+    sourceEndpoint = getProperty(MagixMessage::sourceEndpoint.name).toString(),
+    targetEndpoint = getProperty(MagixMessage::targetEndpoint.name)?.toString(),
+    id = getProperty(MagixMessage::id.name)?.toString(),
+    parentId = getProperty(MagixMessage::parentId.name)?.toString(),
+    user = getBlobString(MagixMessage::user.name)?.let {
+        magixJson.parseToJsonElement(it)
+    },
+)
+
+public class XodusMagixHistory(private val store: PersistentEntityStore) : MagixHistory {
+
+    public fun writeMessage(storeTransaction: StoreTransaction, message: MagixMessage) {
+        storeTransaction.newEntity(XodusMagixStorage.MAGIC_MESSAGE_ENTITY_TYPE).apply {
             setProperty(MagixMessage::sourceEndpoint.name, message.sourceEndpoint)
             setProperty(MagixMessage::format.name, message.format)
 
@@ -59,21 +64,100 @@ public class XodusMagixStorage(
             }
         }
     }
-    }.launchIn(scope)
 
-    private fun Entity.parseMagixMessage(): MagixMessage = MagixMessage(
+    public fun sendMessage(message: MagixMessage) {
-        format = getProperty(MagixMessage::format.name).toString(),
+        store.executeInTransaction { transaction ->
-        payload = getBlobString(MagixMessage::payload.name)?.let {
+            writeMessage(transaction, message)
-            magixJson.parseToJsonElement(it)
+        }
-        } ?: JsonObject(emptyMap()),
+    }
-        sourceEndpoint = getProperty(MagixMessage::sourceEndpoint.name).toString(),
+
-        targetEndpoint = getProperty(MagixMessage::targetEndpoint.name)?.toString(),
+    override suspend fun useMessages(
-        id = getProperty(MagixMessage::id.name)?.toString(),
+        magixFilter: MagixMessageFilter?,
-        parentId = getProperty(MagixMessage::parentId.name)?.toString(),
+        payloadFilter: MagixPayloadFilter?,
-        user = getBlobString(MagixMessage::user.name)?.let {
+        userFilter: MagixUsernameFilter?,
-            magixJson.parseToJsonElement(it)
+        callback: (Sequence<MagixMessage>) -> Unit,
-        },
+    ): Unit = store.executeInReadonlyTransaction { transaction ->
+        val all = transaction.getAll(XodusMagixStorage.MAGIC_MESSAGE_ENTITY_TYPE)
+
+        fun StoreTransaction.findAllIn(
+            entityType: String,
+            field: String,
+            values: Collection<String>?,
+        ): EntityIterable? {
+            var union: EntityIterable? = null
+            values?.forEach {
+                val filter = transaction.find(entityType, field, it)
+                union = union?.union(filter) ?: filter
+            }
+            return union
+        }
+
+        // filter by magix filter
+        val filteredByMagix: EntityIterable = magixFilter?.let { mf ->
+            var res = all
+            transaction.findAllIn(XodusMagixStorage.MAGIC_MESSAGE_ENTITY_TYPE, MagixMessage::format.name, mf.format)
+                ?.let {
+                    res = res.intersect(it)
+                }
+            transaction.findAllIn(
+                XodusMagixStorage.MAGIC_MESSAGE_ENTITY_TYPE,
+                MagixMessage::sourceEndpoint.name,
+                mf.source
+            )?.let {
+                res = res.intersect(it)
+            }
+            transaction.findAllIn(
+                XodusMagixStorage.MAGIC_MESSAGE_ENTITY_TYPE,
+                MagixMessage::targetEndpoint.name,
+                mf.target
+            )?.let {
+                res = res.intersect(it)
+            }
+
+            res
+        } ?: all
+
+        val filteredByUser: EntityIterable = userFilter?.let { userFilter ->
+            filteredByMagix.intersect(
+                transaction.find(
+                    XodusMagixStorage.MAGIC_MESSAGE_ENTITY_TYPE,
+                    MagixMessage::user.name,
+                    userFilter.userName
                 )
+            )
+        } ?: filteredByMagix
+
+
+        val sequence = filteredByUser.asSequence().map { it.parseMagixMessage() }
+
+        val filteredSequence = if (payloadFilter == null) {
+            sequence
+        } else {
+            sequence.filter {
+                payloadFilter.test(it.payload)
+            }
+        }
+
+        callback(filteredSequence)
+    }
+}
+
+/**
+ * Attach a Xodus storage process to the given endpoint.
+ */
+public class XodusMagixStorage(
+    scope: CoroutineScope,
+    private val store: PersistentEntityStore,
+    endpoint: MagixEndpoint,
+    filter: MagixMessageFilter = MagixMessageFilter.ALL,
+) : AutoCloseable {
+
+    public val history: XodusMagixHistory = XodusMagixHistory(store)
+
+    //TODO consider message buffering
+    internal val subscriptionJob = endpoint.subscribe(filter).onEach { message ->
+        history.sendMessage(message)
+    }.launchIn(scope)
 
 
     /**
@@ -105,63 +189,6 @@ public class XodusMagixStorage(
         block(sequence)
     }
 
-    override suspend fun findMessages(
-        magixFilter: MagixMessageFilter?,
-        payloadFilter: MagixPayloadFilter?,
-        userFilter: MagixUsernameFilter?,
-        callback: (Sequence<MagixMessage>) -> Unit,
-    ): Unit = store.executeInReadonlyTransaction { transaction ->
-        val all = transaction.getAll(MAGIC_MESSAGE_ENTITY_TYPE)
-
-        fun StoreTransaction.findAllIn(
-            entityType: String,
-            field: String,
-            values: Collection<String>?,
-        ): EntityIterable? {
-            var union: EntityIterable? = null
-            values?.forEach {
-                val filter = transaction.find(entityType, field, it)
-                union = union?.union(filter) ?: filter
-            }
-            return union
-        }
-
-        // filter by magix filter
-        val filteredByMagix: EntityIterable = magixFilter?.let { mf ->
-            var res = all
-            transaction.findAllIn(MAGIC_MESSAGE_ENTITY_TYPE, MagixMessage::format.name, mf.format)?.let {
-                res = res.intersect(it)
-            }
-            transaction.findAllIn(MAGIC_MESSAGE_ENTITY_TYPE, MagixMessage::sourceEndpoint.name, mf.origin)?.let {
-                res = res.intersect(it)
-            }
-            transaction.findAllIn(MAGIC_MESSAGE_ENTITY_TYPE, MagixMessage::targetEndpoint.name, mf.target)?.let {
-                res = res.intersect(it)
-            }
-
-            res
-        } ?: all
-
-        val filteredByUser: EntityIterable = userFilter?.let { userFilter ->
-            filteredByMagix.intersect(
-                transaction.find(MAGIC_MESSAGE_ENTITY_TYPE, MagixMessage::user.name, userFilter.userName)
-            )
-        } ?: filteredByMagix
-
-
-        val sequence = filteredByUser.asSequence().map { it.parseMagixMessage() }
-
-        val filteredSequence = if (payloadFilter == null) {
-            sequence
-        } else {
-            sequence.filter {
-                payloadFilter.test(it.payload)
-            }
-        }
-
-        callback(filteredSequence)
-    }
-
     override fun close() {
         subscriptionJob.cancel()
     }

magix/magix-storage/magix-storage-xodus/src/main/kotlin/space/kscience/magix/storage/xodus/storageAPITest.kt

@@ -0,0 +1,50 @@
+package space.kscience.magix.storage.xodus
+
+import jetbrains.exodus.entitystore.PersistentEntityStores
+import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.buildJsonObject
+import space.kscience.magix.api.MagixMessage
+import space.kscience.magix.api.MagixMessageFilter
+import java.nio.file.Files
+import kotlin.time.measureTime
+
+public suspend fun main() {
+    val storeDirectory = Files.createTempDirectory("controls-xodus").toFile()
+    println(storeDirectory)
+    val store = PersistentEntityStores.newInstance(storeDirectory)
+    val history = XodusMagixHistory(store)
+
+    store.executeInTransaction { transaction ->
+        for (value in 1..100) {
+            for (source in 1..100) {
+                for (target in 1..100) {
+                    history.writeMessage(
+                        transaction,
+                        MagixMessage(
+                            "test",
+                            sourceEndpoint = "source$source",
+                            targetEndpoint = "target$target",
+                            payload = buildJsonObject {
+                                put("value", JsonPrimitive(value))
+                            }
+                        )
+                    )
+                }
+            }
+        }
+    }
+
+    println("written million messages")
+
+
+    val time = measureTime {
+        history.useMessages(
+            MagixMessageFilter(source = listOf("source12"), target = listOf("target12"))
+        ) { sequence ->
+            println(sequence.count())
+        }
+    }
+    println("Finished query in $time")
+
+    store.close()
+}

magix/magix-storage/src/commonMain/kotlin/space/kscience/magix/storage/MagixHistory.kt

@@ -66,7 +66,7 @@ public interface MagixHistory {
      * @param payloadFilter filter for payload fields.
      * @param userFilter filters user names ("user.name").
      */
-    public suspend fun findMessages(
+    public suspend fun useMessages(
         magixFilter: MagixMessageFilter? = null,
         payloadFilter: MagixPayloadFilter? = null,
         userFilter: MagixUsernameFilter? = null,

magix/magix-storage/src/commonMain/kotlin/space/kscience/magix/storage/historyEndpoint.kt

@@ -57,7 +57,7 @@ public fun MagixEndpoint.launchHistory(
 
     if (payload is HistoryRequestPayload) {
         val realPageSize = payload.pageSize ?: pageSize
-        history.findMessages(payload.magixFilter, payload.payloadFilter, payload.userFilter) { sequence ->
+        history.useMessages(payload.magixFilter, payload.payloadFilter, payload.userFilter) { sequence ->
             // start from -1 because increment always happens first
             var pageNumber = -1
 

magix/magix-zmq/src/main/kotlin/space/kscince/magix/zmq/ZmqMagixEndpoint.kt

@@ -13,6 +13,7 @@ import space.kscience.magix.api.MagixMessage
 import space.kscience.magix.api.MagixMessageFilter
 import space.kscience.magix.api.filter
 import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.coroutineContext
 
 public class ZmqMagixEndpoint(
     private val host: String,
@@ -69,8 +70,7 @@ public class ZmqMagixEndpoint(
     }
 }
 
-public fun MagixEndpoint.Companion.zmq(
+public suspend fun MagixEndpoint.Companion.zmq(
-    scope: CoroutineScope,
     host: String,
     protocol: String = "tcp",
     pubPort: Int = DEFAULT_MAGIX_ZMQ_PUB_PORT,
@@ -80,5 +80,5 @@ public fun MagixEndpoint.Companion.zmq(
     protocol,
     pubPort,
     pullPort,
-    coroutineContext = scope.coroutineContext
+    coroutineContext = coroutineContext
 )

magix/magix-zmq/src/main/kotlin/space/kscince/magix/zmq/ZmqMagixFlowPlugin.kt

@@ -18,6 +18,7 @@ public class ZmqMagixFlowPlugin(
     public val localHost: String = "tcp://*",
     public val zmqPubSocketPort: Int = MagixEndpoint.DEFAULT_MAGIX_ZMQ_PUB_PORT,
     public val zmqPullSocketPort: Int = MagixEndpoint.DEFAULT_MAGIX_ZMQ_PULL_PORT,
+    private val zContext: ZContext = ZContext()
 ) : MagixFlowPlugin {
 
     override fun start(
@@ -27,7 +28,7 @@ public class ZmqMagixFlowPlugin(
     ): Job = scope.launch(Dispatchers.IO) {
         val logger = LoggerFactory.getLogger("magix-server-zmq")
 
-        ZContext().use { context ->
+        zContext.use { context ->
             //launch the publishing job
             val pubSocket = context.createSocket(SocketType.PUB)
             pubSocket.bind("$localHost:$zmqPubSocketPort")