kscience/controls-kt
8
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases 1 Wiki Activity

Refactor xodus storage for local history

Browse Source
This commit is contained in:
Alexander Nozik 2022-06-02 09:21:07 +03:00
parent b6f3769529
commit eb7507191e
No known key found for this signature in database
GPG Key ID: F7FCF2DD25C71357
4 changed files with 122 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
controls-storage/src/jvmMain/kotlin/ru/mipt/npm/controls/storage/storageJvm.kt
View File

@ -1,46 +1,46 @@
package ru.mipt.npm.controls.storage //package ru.mipt.npm.controls.storage
//
import io.ktor.server.application.Application //import io.ktor.server.application.Application
import kotlinx.coroutines.InternalCoroutinesApi //import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.flow.Flow //import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow //import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.filter //import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.onEach //import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.job //import kotlinx.coroutines.job
import ru.mipt.npm.magix.server.GenericMagixMessage //import ru.mipt.npm.magix.server.GenericMagixMessage
import space.kscience.dataforge.context.Factory //import space.kscience.dataforge.context.Factory
import space.kscience.dataforge.meta.Meta //import space.kscience.dataforge.meta.Meta
//
/** ///**
* Asynchronous version of synchronous API, so for more details check relative docs // * Asynchronous version of synchronous API, so for more details check relative docs
*/ // */
//
internal fun Flow<GenericMagixMessage>.store( //internal fun Flow<GenericMagixMessage>.store(
client: EventStorage, // client: EventStorage,
flowFilter: suspend (GenericMagixMessage) -> Boolean = { true }, // flowFilter: suspend (GenericMagixMessage) -> Boolean = { true },
) { //) {
filter(flowFilter).onEach { message -> // filter(flowFilter).onEach { message ->
client.storeMagixMessage(message) // client.storeMagixMessage(message)
} // }
} //}
//
/** Begin to store MagixMessages from certain flow ///** Begin to store MagixMessages from certain flow
* @param flow flow of messages which we will store // * @param flow flow of messages which we will store
* @param meta Meta which may have some configuration parameters for our storage and will be used in invoke method of factory // * @param meta Meta which may have some configuration parameters for our storage and will be used in invoke method of factory
* @param factory factory that will be used for creating persistent entity store instance. DefaultPersistentStoreFactory by default. // * @param factory factory that will be used for creating persistent entity store instance. DefaultPersistentStoreFactory by default.
* @param flowFilter allow you to specify messages which we want to store. Always true by default. // * @param flowFilter allow you to specify messages which we want to store. Always true by default.
*/ // */
@OptIn(InternalCoroutinesApi::class) //@OptIn(InternalCoroutinesApi::class)
public fun Application.store( //public fun Application.store(
flow: MutableSharedFlow<GenericMagixMessage>, // flow: MutableSharedFlow<GenericMagixMessage>,
factory: Factory<EventStorage>, // factory: Factory<EventStorage>,
meta: Meta = Meta.EMPTY, // meta: Meta = Meta.EMPTY,
flowFilter: suspend (GenericMagixMessage) -> Boolean = { true }, // flowFilter: suspend (GenericMagixMessage) -> Boolean = { true },
) { //) {
val client = factory(meta) // val client = factory(meta)
//
flow.store(client, flowFilter) // flow.store(client, flowFilter)
coroutineContext.job.invokeOnCompletion(onCancelling = true) { // coroutineContext.job.invokeOnCompletion(onCancelling = true) {
client.close() // client.close()
} // }
} //}

26
controls-xodus/src/main/kotlin/ru/mipt/npm/controls/xodus/XodusDeviceMessageStorage.kt
View File

@ -27,7 +27,7 @@ import space.kscience.dataforge.names.matches
import space.kscience.dataforge.names.parseAsName import space.kscience.dataforge.names.parseAsName
internal fun StoreTransaction.writeMessage(message: DeviceMessage): Entity { internal fun StoreTransaction.writeMessage(message: DeviceMessage): Unit {
val entity: Entity = newEntity(XodusDeviceMessageStorage.DEVICE_MESSAGE_ENTITY_TYPE) val entity: Entity = newEntity(XodusDeviceMessageStorage.DEVICE_MESSAGE_ENTITY_TYPE)
val json = Json.encodeToJsonElement(DeviceMessage.serializer(), message).jsonObject val json = Json.encodeToJsonElement(DeviceMessage.serializer(), message).jsonObject
val type = json["type"]?.jsonPrimitive?.content ?: error("Message json representation must have type.") val type = json["type"]?.jsonPrimitive?.content ?: error("Message json representation must have type.")
@ -43,8 +43,6 @@ internal fun StoreTransaction.writeMessage(message: DeviceMessage): Entity {
entity.setProperty(DeviceMessage::targetDevice.name, it.toString()) entity.setProperty(DeviceMessage::targetDevice.name, it.toString())
} }
entity.setBlobString("json", Json.encodeToString(json)) entity.setBlobString("json", Json.encodeToString(json))
return entity
} }
@ -65,15 +63,16 @@ public class XodusDeviceMessageStorage(
) : DeviceMessageStorage, AutoCloseable { ) : DeviceMessageStorage, AutoCloseable {
override suspend fun write(event: DeviceMessage) { override suspend fun write(event: DeviceMessage) {
//entityStore.encodeToEntity(event, DEVICE_MESSAGE_ENTITY_TYPE, DeviceMessage.serializer()) entityStore.executeInTransaction { txn ->
entityStore.computeInTransaction { txn ->
txn.writeMessage(event) txn.writeMessage(event)
} }
} }
override suspend fun readAll(): List<DeviceMessage> = entityStore.computeInTransaction { transaction -> override suspend fun readAll(): List<DeviceMessage> = entityStore.computeInReadonlyTransaction { transaction ->
transaction.getAll( transaction.sort(
DEVICE_MESSAGE_ENTITY_TYPE, DEVICE_MESSAGE_ENTITY_TYPE,
DeviceMessage::time.name,
true
).map { ).map {
Json.decodeFromString( Json.decodeFromString(
DeviceMessage.serializer(), DeviceMessage.serializer(),
@ -87,22 +86,21 @@ public class XodusDeviceMessageStorage(
range: ClosedRange<Instant>?, range: ClosedRange<Instant>?,
sourceDevice: Name?, sourceDevice: Name?,
targetDevice: Name?, targetDevice: Name?,
): List<DeviceMessage> = entityStore.computeInTransaction { transaction -> ): List<DeviceMessage> = entityStore.computeInReadonlyTransaction { transaction ->
transaction.find( transaction.find(
DEVICE_MESSAGE_ENTITY_TYPE, DEVICE_MESSAGE_ENTITY_TYPE,
"type", "type",
eventType eventType
).mapNotNull { ).asSequence().filter {
if (it.timeInRange(range) && it.timeInRange(range) &&
it.propertyMatchesName(DeviceMessage::sourceDevice.name, sourceDevice) && it.propertyMatchesName(DeviceMessage::sourceDevice.name, sourceDevice) &&
it.propertyMatchesName(DeviceMessage::targetDevice.name, targetDevice) it.propertyMatchesName(DeviceMessage::targetDevice.name, targetDevice)
) { }.map {
Json.decodeFromString( Json.decodeFromString(
DeviceMessage.serializer(), DeviceMessage.serializer(),
it.getBlobString("json") ?: error("No json content found") it.getBlobString("json") ?: error("No json content found")
) )
} else null }.sortedBy { it.time }.toList()
}
} }
override fun close() { override fun close() {
@ -110,7 +108,7 @@ public class XodusDeviceMessageStorage(
} }
public companion object : Factory<XodusDeviceMessageStorage> { public companion object : Factory<XodusDeviceMessageStorage> {
internal const val DEVICE_MESSAGE_ENTITY_TYPE = "DeviceMessage" internal const val DEVICE_MESSAGE_ENTITY_TYPE = "controls-kt.message"
public val XODUS_STORE_PROPERTY: Name = Name.of("xodus", "storagePath") public val XODUS_STORE_PROPERTY: Name = Name.of("xodus", "storagePath")

39
controls-xodus/src/test/kotlin/PropertyHistoryTest.kt
View File

@ -6,20 +6,19 @@ import org.junit.jupiter.api.AfterAll
import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.BeforeAll import org.junit.jupiter.api.BeforeAll
import org.junit.jupiter.api.Test import org.junit.jupiter.api.Test
import ru.mipt.npm.controls.api.DeviceMessage
import ru.mipt.npm.controls.api.PropertyChangedMessage import ru.mipt.npm.controls.api.PropertyChangedMessage
import ru.mipt.npm.controls.storage.getPropertyHistory import ru.mipt.npm.controls.xodus.XodusDeviceMessageStorage
import ru.mipt.npm.controls.xodus.XODUS_STORE_PROPERTY import ru.mipt.npm.controls.xodus.query
import ru.mipt.npm.controls.xodus.XodusEventStorage import ru.mipt.npm.controls.xodus.writeMessage
import ru.mipt.npm.xodus.serialization.json.encodeToEntity
import space.kscience.dataforge.meta.Meta import space.kscience.dataforge.meta.Meta
import space.kscience.dataforge.names.Name import space.kscience.dataforge.names.Name
import java.io.File import space.kscience.dataforge.names.asName
import java.nio.file.Files
internal class PropertyHistoryTest { internal class PropertyHistoryTest {
companion object { companion object {
private val storeName = ".property_history_test" val storeFile = Files.createTempDirectory("controls-xodus").toFile()
private val entityStore = PersistentEntityStores.newInstance(storeName)
private val propertyChangedMessages = listOf( private val propertyChangedMessages = listOf(
PropertyChangedMessage( PropertyChangedMessage(
@ -45,28 +44,34 @@ internal class PropertyHistoryTest {
@BeforeAll @BeforeAll
@JvmStatic @JvmStatic
fun createEntities() { fun createEntities() {
propertyChangedMessages.forEach { PersistentEntityStores.newInstance(storeFile).use {
entityStore.encodeToEntity<DeviceMessage>(it, "DeviceMessage") it.executeInTransaction { transaction ->
propertyChangedMessages.forEach { message ->
transaction.writeMessage(message)
}
}
} }
entityStore.close()
} }
@AfterAll @AfterAll
@JvmStatic @JvmStatic
fun deleteDatabase() { fun deleteDatabase() {
File(storeName).deleteRecursively() storeFile.deleteRecursively()
} }
} }
@OptIn(ExperimentalCoroutinesApi::class) @OptIn(ExperimentalCoroutinesApi::class)
@Test @Test
fun getPropertyHistoryTest() = runTest { fun getPropertyHistoryTest() = runTest {
PersistentEntityStores.newInstance(storeFile).use { entityStore ->
XodusDeviceMessageStorage(entityStore).use { storage ->
assertEquals( assertEquals(
listOf(propertyChangedMessages[0]), propertyChangedMessages[0],
getPropertyHistory( storage.query<PropertyChangedMessage>(
"virtual-car", "speed", XodusEventStorage, Meta { sourceDevice = "virtual-car".asName()
XODUS_STORE_PROPERTY put storeName ).first { it.property == "speed" }
})
) )
} }
}
}
} }

41
magix/magix-storage/magix-storage-xodus/src/main/kotlin/ru/mipt/npm/magix/storage/xodus/XodusMagixStorage.kt
View File

@ -1,25 +1,52 @@
package ru.mipt.npm.magix.storage.xodus package ru.mipt.npm.magix.storage.xodus
import jetbrains.exodus.entitystore.PersistentEntityStore
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.onEach
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.JsonElement import kotlinx.serialization.json.JsonElement
import ru.mipt.npm.magix.api.MagixEndpoint import ru.mipt.npm.magix.api.MagixEndpoint
import ru.mipt.npm.magix.api.MagixMessage
import ru.mipt.npm.magix.api.MagixMessageFilter import ru.mipt.npm.magix.api.MagixMessageFilter
import java.nio.file.Path
public class XodusMagixStorage( public class XodusMagixStorage(
private val scope: CoroutineScope, scope: CoroutineScope,
private val path: Path, private val store: PersistentEntityStore,
private val endpoint: MagixEndpoint<JsonElement>, endpoint: MagixEndpoint<JsonElement>,
private val filter: MagixMessageFilter = MagixMessageFilter(), filter: MagixMessageFilter = MagixMessageFilter(),
) : AutoCloseable { ) : AutoCloseable {
private val subscriptionJob = endpoint.subscribe(filter).onEach { //TODO consider message buffering
TODO() private val subscriptionJob = endpoint.subscribe(filter).onEach { message ->
store.executeInTransaction { transaction ->
transaction.newEntity(MAGIC_MESSAGE_ENTITY_TYPE).apply {
setProperty(MagixMessage<*>::origin.name, message.origin)
setProperty(MagixMessage<*>::format.name, message.format)
setBlobString(MagixMessage<*>::payload.name, MagixEndpoint.magixJson.encodeToString(message.payload))
message.target?.let {
setProperty(MagixMessage<*>::target.name, it)
}
message.id?.let {
setProperty(MagixMessage<*>::id.name, it)
}
message.parentId?.let {
setProperty(MagixMessage<*>::parentId.name, it)
}
message.user?.let {
setBlobString(MagixMessage<*>::user.name, MagixEndpoint.magixJson.encodeToString(it))
}
}
}
}.launchIn(scope) }.launchIn(scope)
override fun close() { override fun close() {
subscriptionJob.cancel() subscriptionJob.cancel()
} }
public companion object {
public const val MAGIC_MESSAGE_ENTITY_TYPE: String = "magix.message"
}
} }