Implement MagixVirtualCar

demo/car/build.gradle.kts

@@ -20,6 +20,7 @@ dependencies {
     implementation(projects.magix.magixRsocket)
     implementation(projects.controlsMagixClient)
 
+    implementation("io.ktor:ktor-client-cio:$ktorVersion")
     implementation("org.jetbrains.kotlinx:kotlinx-datetime:0.3.1")
     implementation("no.tornado:tornadofx:1.7.20")
     implementation("space.kscience:plotlykt-server:0.5.0-dev-1")

demo/car/src/main/kotlin/ru/mipt/npm/controls/demo/car/MagixVirtualCar.kt

@@ -0,0 +1,58 @@
+package ru.mipt.npm.controls.demo.car
+
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
+import ru.mipt.npm.controls.api.DeviceMessage
+import ru.mipt.npm.controls.api.PropertyChangedMessage
+import ru.mipt.npm.controls.spec.DeviceBySpec
+import ru.mipt.npm.controls.spec.doRecurring
+import ru.mipt.npm.magix.api.MagixEndpoint
+import space.kscience.dataforge.context.Context
+import space.kscience.dataforge.context.Factory
+import space.kscience.dataforge.meta.Meta
+import space.kscience.dataforge.names.Name
+import kotlin.time.Duration
+import kotlin.time.ExperimentalTime
+
+class MagixVirtualCar(private val magixEndpoint: MagixEndpoint<DeviceMessage>, context: Context, meta: Meta)
+    : DeviceBySpec<MagixVirtualCar>(IVirtualCar, context, meta), IVirtualCar {
+    override var speedState: Vector2D = Vector2D()
+    override var locationState: Vector2D = Vector2D()
+    override var accelerationState: Vector2D = Vector2D()
+
+    private suspend fun MagixEndpoint<DeviceMessage>.startMagixVirtualCarUpdate() {
+        launch {
+            subscribe().collect { magix ->
+                (magix.payload as? PropertyChangedMessage)?.let { message ->
+                    if (message.sourceDevice == Name.parse("virtual-car")) {
+                        when (message.property) {
+                            "speed" -> speedState = Vector2D.metaToObject(message.value)
+                            "location" -> locationState = Vector2D.metaToObject(message.value)
+                            "acceleration" -> accelerationState = Vector2D.metaToObject(message.value)
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    @OptIn(ExperimentalTime::class)
+    override suspend fun open() {
+        super<DeviceBySpec>.open()
+
+        launch {
+            magixEndpoint.startMagixVirtualCarUpdate()
+        }
+
+        //starting regular updates
+        doRecurring(Duration.milliseconds(100)) {
+            IVirtualCar.speed.read()
+            IVirtualCar.location.read()
+            IVirtualCar.acceleration.read()
+        }
+    }
+}
+
+class MagixVirtualCarFactory(private val magixEndpoint: MagixEndpoint<DeviceMessage>) : Factory<MagixVirtualCar> {
+    override fun invoke(meta: Meta, context: Context): MagixVirtualCar = MagixVirtualCar(magixEndpoint, context, meta)
+}

demo/car/src/main/kotlin/ru/mipt/npm/controls/demo/car/VirtualCarController.kt

@@ -14,13 +14,15 @@ import ru.mipt.npm.controls.controllers.install
 import ru.mipt.npm.controls.demo.car.IVirtualCar.Companion.acceleration
 import ru.mipt.npm.magix.api.MagixEndpoint
 import ru.mipt.npm.magix.rsocket.rSocketWithTcp
+import ru.mipt.npm.magix.rsocket.rSocketWithWebSockets
 import ru.mipt.npm.magix.server.startMagixServer
 import space.kscience.dataforge.context.*
 import tornadofx.*
 
 class VirtualCarController : Controller(), ContextAware {
 
-    var device: VirtualCar? = null
+    var virtualCar: VirtualCar? = null
+    var magixVirtualCar: MagixVirtualCar? = null
     var magixServer: ApplicationEngine? = null
 
     override val context = Context("demoDevice") {
@@ -31,9 +33,11 @@ class VirtualCarController : Controller(), ContextAware {
 
     fun init() {
         context.launch {
-            device = deviceManager.install("virtual-car", VirtualCar)
+            virtualCar = deviceManager.install("virtual-car", VirtualCar)
             //starting magix event loop
             magixServer = startMagixServer(enableRawRSocket = true, enableZmq = true)
+            val magixEndpoint = MagixEndpoint.rSocketWithWebSockets("localhost", DeviceMessage.serializer())
+            magixVirtualCar = deviceManager.install("magix-virtual-car", MagixVirtualCarFactory(magixEndpoint))
             //Launch device client and connect it to the server
             val deviceEndpoint = MagixEndpoint.rSocketWithTcp("localhost", DeviceMessage.serializer())
             deviceManager.connectToMagix(deviceEndpoint)
@@ -44,8 +48,10 @@ class VirtualCarController : Controller(), ContextAware {
         logger.info { "Shutting down..." }
         magixServer?.stop(1000, 5000)
         logger.info { "Magix server stopped" }
-        device?.close()
+        magixVirtualCar?.close()
-        logger.info { "Device server stopped" }
+        logger.info { "Magix virtual car server stopped" }
+        virtualCar?.close()
+        logger.info { "Virtual car server stopped" }
         context.close()
     }
 }
@@ -78,7 +84,7 @@ class VirtualCarControllerView : View(title = " Virtual car controller remote")
         button("Submit") {
             useMaxWidth = true
             action {
-                controller.device?.run {
+                controller.virtualCar?.run {
                     launch {
                         acceleration.write(Vector2D(accelerationXProperty.get(),
                             accelerationYProperty.get()))

magix/magix-api/src/commonMain/kotlin/ru/mipt/npm/magix/api/MagixMessageFilter.kt

@@ -24,7 +24,6 @@ public fun <T> Flow<MagixMessage<T>>.filter(filter: MagixMessageFilter): Flow<Ma
     }
     return filter { message ->
         filter.format?.contains(message.format) ?: true
-                && filter.origin?.contains(message.origin) ?: true
                 && filter.origin?.contains(message.origin) ?: true
                 && filter.target?.contains(message.target) ?: true
     }