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

Dev #8

Merged
altavir merged 78 commits from dev into master 2023-03-04 16:47:55 +03:00
Conversation 0 Commits 78 Files Changed 137 +14 -48
2 changed files with 14 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 1d8cc33c91 - Show all commits

7
controls-magix-client/src/commonMain/kotlin/ru/mipt/npm/controls/client/dfMagix.kt
View File

@ -29,6 +29,7 @@ internal fun generateId(request: MagixMessage<*>): String = if (request.id != nu
public fun DeviceManager.connectToMagix( public fun DeviceManager.connectToMagix(
endpoint: MagixEndpoint<DeviceMessage>, endpoint: MagixEndpoint<DeviceMessage>,
endpointID: String = DATAFORGE_MAGIX_FORMAT, endpointID: String = DATAFORGE_MAGIX_FORMAT,
preSendAction: (MagixMessage<*>) -> Unit = {}
): Job = context.launch { ): Job = context.launch {
endpoint.subscribe().onEach { request -> endpoint.subscribe().onEach { request ->
val responsePayload = respondHubMessage(request.payload) val responsePayload = respondHubMessage(request.payload)
@ -48,13 +49,15 @@ public fun DeviceManager.connectToMagix(
}.launchIn(this) }.launchIn(this)
hubMessageFlow(this).onEach { payload -> hubMessageFlow(this).onEach { payload ->
endpoint.broadcast( val magixMessage = MagixMessage(
MagixMessage(
format = DATAFORGE_MAGIX_FORMAT, format = DATAFORGE_MAGIX_FORMAT,
id = "df[${payload.hashCode()}]", id = "df[${payload.hashCode()}]",
origin = endpointID, origin = endpointID,
payload = payload payload = payload
) )
preSendAction(magixMessage)
endpoint.broadcast(
magixMessage
) )
}.catch { error -> }.catch { error ->
logger.error(error) { "Error while sending a message" } logger.error(error) { "Error while sending a message" }

43
controls-magix-client/src/jvmMain/kotlin/ru/mipt/npm/controls/client/xodusDfMagix.kt
View File

@ -2,20 +2,11 @@ package ru.mipt.npm.controls.client
import jetbrains.exodus.entitystore.PersistentEntityStore import jetbrains.exodus.entitystore.PersistentEntityStore
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import ru.mipt.npm.controls.api.DeviceMessage 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.controllers.DeviceManager import ru.mipt.npm.controls.controllers.DeviceManager
import ru.mipt.npm.controls.controllers.hubMessageFlow
import ru.mipt.npm.controls.controllers.respondHubMessage
import ru.mipt.npm.controls.xodus.toEntity import ru.mipt.npm.controls.xodus.toEntity
import ru.mipt.npm.magix.api.MagixEndpoint import ru.mipt.npm.magix.api.MagixEndpoint
import ru.mipt.npm.magix.api.MagixMessage
import space.kscience.dataforge.context.error
import space.kscience.dataforge.context.logger
/** /**
* Communicate with server in [Magix format](https://github.com/waltz-controls/rfc/tree/master/1) and dump messages at xodus entity store * Communicate with server in [Magix format](https://github.com/waltz-controls/rfc/tree/master/1) and dump messages at xodus entity store
@ -24,38 +15,10 @@ public fun DeviceManager.connectToMagix(
endpoint: MagixEndpoint<DeviceMessage>, endpoint: MagixEndpoint<DeviceMessage>,
endpointID: String = DATAFORGE_MAGIX_FORMAT, endpointID: String = DATAFORGE_MAGIX_FORMAT,
entityStore: PersistentEntityStore entityStore: PersistentEntityStore
): Job = context.launch { ): Job = connectToMagix(endpoint, endpointID) { message ->
endpoint.subscribe().onEach { request -> if (message.payload is PropertyChangedMessage) {
val responsePayload = respondHubMessage(request.payload)
if (responsePayload != null) {
val response = MagixMessage(
format = DATAFORGE_MAGIX_FORMAT,
id = generateId(request),
parentId = request.id,
origin = endpointID,
payload = responsePayload
)
endpoint.broadcast(response)
}
}.catch { error ->
logger.error(error) { "Error while responding to message" }
}.launchIn(this)
hubMessageFlow(this).onEach { payload ->
val magixMessage = MagixMessage(
format = DATAFORGE_MAGIX_FORMAT,
id = "df[${payload.hashCode()}]",
origin = endpointID,
payload = payload
)
endpoint.broadcast(magixMessage)
if (payload is PropertyChangedMessage) {
entityStore.executeInTransaction { entityStore.executeInTransaction {
magixMessage.toEntity(it) message.toEntity(it)
} }
} }
}.catch { error ->
logger.error(error) { "Error while sending a message" }
}.launchIn(this)
} }