controls-constructor/src/commonMain/kotlin/space/kscience/controls/constructor/DeviceState.kt

@@ -74,16 +74,15 @@ public fun <T, R> DeviceState.Companion.map(
 
 public fun <T, R> DeviceState<T>.map(mapper: (T) -> R): DeviceStateWithDependencies<R> = DeviceState.map(this, mapper)
 
-public fun DeviceState<NumericalValue<out UnitsOfMeasurement>>.values(): DeviceState<Double> =
+public fun DeviceState<NumericalValue<out UnitsOfMeasurement>>.values(): DeviceState<Double> = object : DeviceState<Double> {
-    object : DeviceState<Double> {
+    override val value: Double
-        override val value: Double
+        get() = this@values.value.value
-            get() = this@values.value.value
 
-        override val valueFlow: Flow<Double>
+    override val valueFlow: Flow<Double>
-            get() = this@values.valueFlow.map { it.value }
+        get() = this@values.valueFlow.map { it.value }
 
-        override fun toString(): String = this@values.toString()
+    override fun toString(): String = this@values.toString()
-    }
+}
 
 /**
  * Combine two device states into one read-only [DeviceState]. Only the latest value of each state is used.

controls-core/src/commonMain/kotlin/space/kscience/controls/misc/converters.kt

@@ -41,10 +41,9 @@ private object InstantConverter : MetaConverter<Instant> {
 public val MetaConverter.Companion.instant: MetaConverter<Instant> get() = InstantConverter
 
 private object DoubleRangeConverter : MetaConverter<ClosedFloatingPointRange<Double>> {
-    override fun readOrNull(source: Meta): ClosedFloatingPointRange<Double>? =
+    override fun readOrNull(source: Meta): ClosedFloatingPointRange<Double>? = source.value?.doubleArray?.let { (start, end)->
-        source.value?.doubleArray?.let { (start, end) ->
+        start..end
-            start..end
+    }
-        }
 
     override fun convert(
         obj: ClosedFloatingPointRange<Double>,
@@ -52,11 +51,3 @@ private object DoubleRangeConverter : MetaConverter<ClosedFloatingPointRange<Dou
 }
 
 public val MetaConverter.Companion.doubleRange: MetaConverter<ClosedFloatingPointRange<Double>> get() = DoubleRangeConverter
-
-private object StringListConverter : MetaConverter<List<String>> {
-    override fun convert(obj: List<String>): Meta = Meta(obj.map { it.asValue() }.asValue())
-
-    override fun readOrNull(source: Meta): List<String>? = source.stringList ?: source["@jsonArray"]?.stringList
-}
-
-public val MetaConverter.Companion.stringList: MetaConverter<List<String>> get() = StringListConverter

controls-core/src/commonMain/kotlin/space/kscience/controls/peer/PeerConnection.kt

@@ -2,13 +2,14 @@ package space.kscience.controls.peer
 
 import space.kscience.dataforge.io.Envelope
 import space.kscience.dataforge.meta.Meta
+import space.kscience.dataforge.names.Name
 
 /**
  * A manager that allows direct synchronous sending and receiving binary data
  */
 public interface PeerConnection {
     /**
-     * Receive an [Envelope] from a device on a given [address] with given [contentId].
+     * Receive an [Envelope] from a device with name [deviceName] on a given [address] with given [contentId].
      *
      * The address depends on the specifics of given [PeerConnection]. For example, it could be a TCP/IP port or
      * magix endpoint name.
@@ -19,12 +20,13 @@ public interface PeerConnection {
      */
     public suspend fun receive(
         address: String,
+        deviceName: Name,
         contentId: String,
         requestMeta: Meta = Meta.EMPTY,
     ): Envelope
 
     /**
-     * Send an [envelope] to a device on a given [address]
+     * Send an [envelope] to a device with name [deviceName] on a given [address]
      *
      * The address depends on the specifics of given [PeerConnection]. For example, it could be a TCP/IP port or
      * magix endpoint name.
@@ -33,6 +35,7 @@ public interface PeerConnection {
      */
     public suspend fun send(
         address: String,
+        deviceName: Name,
         envelope: Envelope,
         requestMeta: Meta = Meta.EMPTY,
     )

controls-magix/src/commonMain/kotlin/space/kscience/controls/client/clientPropertyAccess.kt

@@ -19,13 +19,10 @@ import space.kscience.dataforge.meta.Meta
 public suspend fun <T> DeviceClient.read(propertySpec: DevicePropertySpec<*, T>): T =
     propertySpec.converter.readOrNull(readProperty(propertySpec.name)) ?: error("Property read result is not valid")
 
+
 public suspend fun <T> DeviceClient.request(propertySpec: DevicePropertySpec<*, T>): T =
     propertySpec.converter.read(getOrReadProperty(propertySpec.name))
 
-public fun <T> DeviceClient.getCached(propertySpec: DevicePropertySpec<*, T>): T? =
-    getProperty(propertySpec.name)?.let { propertySpec.converter.read(it) }
-
-
 public suspend fun <T> DeviceClient.write(propertySpec: MutableDevicePropertySpec<*, T>, value: T) {
     writeProperty(propertySpec.name, propertySpec.converter.convert(value))
 }

controls-server/src/jvmMain/kotlin/space/kscience/controls/server/deviceWebServer.kt

@@ -38,7 +38,6 @@ import space.kscience.dataforge.names.get
 import space.kscience.magix.api.MagixEndpoint
 import space.kscience.magix.api.MagixFlowPlugin
 import space.kscience.magix.api.MagixMessage
-import space.kscience.magix.api.start
 import space.kscience.magix.server.magixModule
 
 
@@ -216,6 +215,5 @@ public fun Application.deviceManagerModule(
     plugins.forEach {
         it.start(this, magixFlow)
     }
-
     magixModule(magixFlow)
 }

demo/device-collective/build.gradle.kts

@@ -13,9 +13,7 @@ kscience {
     commonMain {
         implementation(projects.controlsVisualisationCompose)
         implementation(projects.controlsConstructor)
-        implementation(projects.magix.magixServer)
         implementation(projects.magix.magixRsocket)
-        implementation(projects.controlsMagix)
     }
     jvmMain {
 //        implementation("io.ktor:ktor-server-cio")

demo/device-collective/src/jvmMain/kotlin/CollectiveDevice.kt

@@ -3,36 +3,27 @@
 package space.kscience.controls.demo.collective
 
 import space.kscience.controls.api.Device
-import space.kscience.controls.constructor.*
+import space.kscience.controls.constructor.DeviceConstructor
-import space.kscience.controls.misc.stringList
+import space.kscience.controls.constructor.MutableDeviceState
-import space.kscience.controls.peer.PeerConnection
+import space.kscience.controls.constructor.registerAsProperty
 import space.kscience.controls.spec.DeviceSpec
 import space.kscience.dataforge.context.Context
 import space.kscience.dataforge.meta.MetaConverter
 import space.kscience.dataforge.meta.Scheme
-import space.kscience.dataforge.meta.int
 import space.kscience.dataforge.meta.string
 import space.kscience.dataforge.misc.DFExperimental
 import space.kscience.maps.coordinates.Gmc
-import space.kscience.maps.coordinates.GmcCurve
 import kotlin.time.Duration.Companion.milliseconds
 
-typealias CollectiveDeviceId = String
+class CollectiveDeviceConfiguration(deviceId: DeviceId) : Scheme() {
-
-class CollectiveDeviceConfiguration(deviceId: CollectiveDeviceId) : Scheme() {
     var deviceId by string(deviceId)
     var description by string()
-    var reportInterval by int(500)
-    var radioFrequency by string(default = "169 MHz")
 }
 
-typealias CollectiveDeviceRoster = Map<CollectiveDeviceId, CollectiveDeviceConfiguration>
 
 interface CollectiveDevice : Device {
 
-    public val id: CollectiveDeviceId
+    public val id: DeviceId
-
-    public val peerConnection: PeerConnection
 
     suspend fun getPosition(): Gmc
 
@@ -40,7 +31,8 @@ interface CollectiveDevice : Device {
 
     suspend fun setVelocity(value: GmcVelocity)
 
-    suspend fun listVisible(): Collection<CollectiveDeviceId>
+    suspend fun listVisible(): Collection<DeviceId>
+
 
     companion object : DeviceSpec<CollectiveDevice>() {
         val position by property<Gmc>(
@@ -53,17 +45,6 @@ interface CollectiveDevice : Device {
             read = { getVelocity() },
             write = { _, value -> setVelocity(value) }
         )
-
-        val visibleNeighbors by property(
-            MetaConverter.stringList,
-            read = {
-                listVisible().toList()
-            }
-        )
-
-//        val listVisible by action(MetaConverter.unit, MetaConverter.valueList<String> { it.string }) {
-//            listVisible().toList()
-//        }
     }
 }
 
@@ -73,34 +54,13 @@ class CollectiveDeviceConstructor(
     val configuration: CollectiveDeviceConfiguration,
     position: MutableDeviceState<Gmc>,
     velocity: MutableDeviceState<GmcVelocity>,
-    override val peerConnection: PeerConnection,
+    private val listVisible: suspend () -> Collection<DeviceId>,
-    private val observation: suspend () -> Map<CollectiveDeviceId, GmcCurve>,
 ) : DeviceConstructor(context, configuration.meta), CollectiveDevice {
 
-    override val id: CollectiveDeviceId get() = configuration.deviceId
+    override val id: DeviceId get() = configuration.deviceId
 
-    val position = registerAsProperty(
+    val position = registerAsProperty(CollectiveDevice.position, position.sample(500.milliseconds))
-        CollectiveDevice.position,
+    val velocity = registerAsProperty(CollectiveDevice.velocity, velocity)
-        position.sample(configuration.reportInterval.milliseconds)
-    )
-
-    val velocity = registerAsProperty(
-        CollectiveDevice.velocity,
-        velocity.sample(configuration.reportInterval.milliseconds)
-    )
-
-    private val _visibleNeighbors: MutableDeviceState<Collection<CollectiveDeviceId>> = stateOf(emptyList())
-
-    val visibleNeighbors = registerAsProperty(
-        CollectiveDevice.visibleNeighbors,
-        _visibleNeighbors.map { it.toList() }
-    )
-
-    init {
-        position.onNext {
-            _visibleNeighbors.value = observation.invoke().keys
-        }
-    }
 
     override suspend fun getPosition(): Gmc = position.value
 
@@ -110,5 +70,5 @@ class CollectiveDeviceConstructor(
         velocity.value = value
     }
 
-    override suspend fun listVisible(): Collection<CollectiveDeviceId> = observation.invoke().keys
+    override suspend fun listVisible(): Collection<DeviceId> = listVisible.invoke()
 }

demo/device-collective/src/jvmMain/kotlin/DeviceCollectiveModel.kt

@@ -1,39 +1,27 @@
 package space.kscience.controls.demo.collective
 
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.launch
-import space.kscience.controls.client.launchMagixService
 import space.kscience.controls.constructor.ModelConstructor
 import space.kscience.controls.constructor.MutableDeviceState
 import space.kscience.controls.constructor.onTimer
-import space.kscience.controls.manager.DeviceManager
-import space.kscience.controls.manager.install
-import space.kscience.controls.peer.PeerConnection
 import space.kscience.dataforge.context.Context
-import space.kscience.dataforge.context.request
-import space.kscience.dataforge.io.Envelope
-import space.kscience.dataforge.meta.Meta
-import space.kscience.dataforge.names.parseAsName
-import space.kscience.magix.api.MagixEndpoint
-import space.kscience.magix.rsocket.rSocketWithWebSockets
-import space.kscience.magix.server.startMagixServer
 import space.kscience.maps.coordinates.*
 
 
-internal data class CollectiveDeviceState(
+typealias DeviceId = String
-    val id: CollectiveDeviceId,
+
+
+internal data class VirtualDeviceState(
+    val id: DeviceId,
     val configuration: CollectiveDeviceConfiguration,
     val position: MutableDeviceState<Gmc>,
     val velocity: MutableDeviceState<GmcVelocity>,
 )
 
 internal fun VirtualDeviceState(
-    id: CollectiveDeviceId,
+    id: DeviceId,
     position: Gmc,
     configuration: CollectiveDeviceConfiguration.() -> Unit = {},
-) = CollectiveDeviceState(
+) = VirtualDeviceState(
     id,
     CollectiveDeviceConfiguration(id).apply(configuration),
     MutableDeviceState(position),
@@ -43,10 +31,9 @@ internal fun VirtualDeviceState(
 
 internal class DeviceCollectiveModel(
     context: Context,
-    val deviceStates: Collection<CollectiveDeviceState>,
+    val deviceStates: Collection<VirtualDeviceState>,
-    val visibilityRange: Distance = 0.5.kilometers,
+    val visibilityRange: Distance,
-    val radioRange: Distance = 5.kilometers,
+) : ModelConstructor(context) {
-) : ModelConstructor(context), PeerConnection {
 
     /**
      * Propagate movement
@@ -58,7 +45,7 @@ internal class DeviceCollectiveModel(
         }
     }
 
-    private fun locateVisible(id: CollectiveDeviceId): Map<CollectiveDeviceId, GmcCurve> {
+    suspend fun locateVisible(id: DeviceId): Map<DeviceId, GmcCurve> {
         val coordinatesSnapshot = deviceStates.associate { it.id to it.position.value }
 
         val selected = coordinatesSnapshot[id] ?: error("Can't find device with id $id")
@@ -69,53 +56,4 @@ internal class DeviceCollectiveModel(
 
         return allCurves.filterValues { it.distance in 0.kilometers..visibilityRange }
     }
-
-    val devices = deviceStates.associate {
-        val device = CollectiveDeviceConstructor(
-            context = context,
-            configuration = it.configuration,
-            position = it.position,
-            velocity = it.velocity,
-            peerConnection = this,
-        ) {
-            locateVisible(it.id)
-        }
-        //start movement program
-        device.moveInCircles()
-        it.id to device
-    }
-
-    val roster = deviceStates.associate { it.id to it.configuration }
-
-    override suspend fun receive(address: String, contentId: String, requestMeta: Meta): Envelope {
-        TODO("Not yet implemented")
-    }
-
-    override suspend fun send(address: String, envelope: Envelope, requestMeta: Meta) {
-//        devices.values.filter { it.configuration.radioFrequency == address }.forEach { device ->
-//            ```
-//        }
-    }
-}
-
-internal fun CoroutineScope.launchCollectiveMagixServer(
-    collectiveModel: DeviceCollectiveModel,
-): Job = launch(Dispatchers.IO) {
-    val server = startMagixServer(
-//        RSocketMagixFlowPlugin()
-    )
-    val deviceEndpoint = MagixEndpoint.rSocketWithWebSockets("localhost")
-
-    collectiveModel.devices.forEach { (id, device) ->
-        val deviceContext = collectiveModel.context.buildContext(id.parseAsName()) {
-            coroutineContext(coroutineContext)
-            plugin(DeviceManager)
-        }
-
-        deviceContext.install(id, device)
-
-//        val deviceEndpoint = MagixEndpoint.rSocketWithWebSockets("localhost")
-
-        deviceContext.request(DeviceManager).launchMagixService(deviceEndpoint, id)
-    }
 }

demo/device-collective/src/jvmMain/kotlin/SampleDeviceState.kt

@@ -4,7 +4,6 @@ import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.sample
 import space.kscience.controls.constructor.DeviceState
-import space.kscience.controls.constructor.MutableDeviceState
 import kotlin.time.Duration
 
 @OptIn(FlowPreview::class)
@@ -12,24 +11,10 @@ class SampleDeviceState<T>(
     val origin: DeviceState<T>,
     val interval: Duration,
 ) : DeviceState<T> {
-    override val value: T by origin::value
+    override val value: T get() = origin.value
     override val valueFlow: Flow<T> get() = origin.valueFlow.sample(interval)
 
     override fun toString(): String = "DebounceDeviceState($value, interval=$interval)"
 }
 
-
 fun <T> DeviceState<T>.sample(interval: Duration) = SampleDeviceState(this, interval)
-
-@OptIn(FlowPreview::class)
-class MutableSampleDeviceState<T>(
-    val origin: MutableDeviceState<T>,
-    val interval: Duration,
-) : MutableDeviceState<T> {
-    override var value: T by origin::value
-    override val valueFlow: Flow<T> get() = origin.valueFlow.sample(interval)
-
-    override fun toString(): String = "DebounceDeviceState($value, interval=$interval)"
-}
-
-fun <T> MutableDeviceState<T>.sample(interval: Duration) = MutableSampleDeviceState(this, interval)

demo/device-collective/src/jvmMain/kotlin/debugModel.kt

@@ -12,8 +12,6 @@ import space.kscience.maps.coordinates.Gmc
 import space.kscience.maps.coordinates.kilometers
 import kotlin.math.PI
 import kotlin.random.Random
-import kotlin.time.Duration
-import kotlin.time.Duration.Companion.milliseconds
 
 private val deviceVelocity = 0.1.kilometers
 
@@ -21,13 +19,8 @@ private val center = Gmc.ofDegrees(55.925, 37.514)
 private val radius = 0.01.degrees
 
 
-internal fun generateModel(
+internal fun generateModel(context: Context): DeviceCollectiveModel {
-    context: Context,
+    val devices: List<VirtualDeviceState> = List(100) { index ->
-    size: Int = 50,
-    reportInterval: Duration = 500.milliseconds,
-    additionalConfiguration: CollectiveDeviceConfiguration.() -> Unit = {},
-): DeviceCollectiveModel {
-    val devices: List<CollectiveDeviceState> = List(size) { index ->
         val id = "device[$index]"
 
         VirtualDeviceState(
@@ -39,12 +32,10 @@ internal fun generateModel(
         ) {
             deviceId = id
             description = "Virtual remote device $id"
-            this.reportInterval = reportInterval.inWholeMilliseconds.toInt()
-            additionalConfiguration()
         }
     }
 
-    val model = DeviceCollectiveModel(context, devices)
+    val model = DeviceCollectiveModel(context, devices, 0.2.kilometers)
 
     return model
 }

demo/device-collective/src/jvmMain/kotlin/main.kt

@@ -10,7 +10,6 @@ import androidx.compose.foundation.layout.padding
 import androidx.compose.material.Card
 import androidx.compose.material.Checkbox
 import androidx.compose.material.Text
-import androidx.compose.material3.CircularProgressIndicator
 import androidx.compose.material3.MaterialTheme
 import androidx.compose.runtime.*
 import androidx.compose.ui.Alignment
@@ -25,103 +24,74 @@ import androidx.compose.ui.window.Window
 import androidx.compose.ui.window.application
 import io.ktor.client.HttpClient
 import io.ktor.client.engine.cio.CIO
-import kotlinx.coroutines.CompletableDeferred
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext
 import org.jetbrains.compose.splitpane.ExperimentalSplitPaneApi
 import org.jetbrains.compose.splitpane.HorizontalSplitPane
 import org.jetbrains.compose.splitpane.rememberSplitPaneState
-import space.kscience.controls.api.PropertyChangedMessage
-import space.kscience.controls.client.*
 import space.kscience.controls.compose.conditional
 import space.kscience.controls.manager.DeviceManager
+import space.kscience.controls.spec.useProperty
 import space.kscience.dataforge.context.Context
-import space.kscience.dataforge.context.ContextBuilder
+import space.kscience.dataforge.context.request
-import space.kscience.dataforge.meta.MetaConverter
-import space.kscience.dataforge.names.parseAsName
-import space.kscience.magix.api.MagixEndpoint
-import space.kscience.magix.api.subscribe
-import space.kscience.magix.rsocket.rSocketWithWebSockets
 import space.kscience.maps.compose.MapView
 import space.kscience.maps.compose.OpenStreetMapTileProvider
-import space.kscience.maps.coordinates.Gmc
-import space.kscience.maps.coordinates.meters
 import space.kscience.maps.features.*
 import java.nio.file.Path
-import kotlin.time.Duration.Companion.seconds
 
 
 @Composable
-fun rememberContext(name: String, contextBuilder: ContextBuilder.() -> Unit = {}): Context = remember {
+fun rememberDeviceManager(): DeviceManager = remember {
-    Context(name, contextBuilder)
+    val context = Context {
-}
+        plugin(DeviceManager)
+    }
 
-private val gmcMetaConverter = MetaConverter.serializable<Gmc>()
+    context.request(DeviceManager)
+}
 
 @Composable
 fun App() {
     val scope = rememberCoroutineScope()
 
-    val parentContext = rememberContext("Parent") {
+
-        plugin(DeviceManager)
+    val deviceManager = rememberDeviceManager()
-    }
+
 
     val collectiveModel = remember {
-        generateModel(parentContext, 100, reportInterval = 1.seconds)
+        generateModel(deviceManager.context)
     }
 
-    val roster = remember {
+    val devices: Map<DeviceId, CollectiveDevice> = remember {
-        collectiveModel.roster
+        collectiveModel.deviceStates.associate {
-    }
+            val device = CollectiveDeviceConstructor(
-
+                context = deviceManager.context,
-    val client = remember { CompletableDeferred<MagixEndpoint>() }
+                configuration = it.configuration,
-
+                position = it.position,
-    val devices = remember { mutableStateMapOf<CollectiveDeviceId, DeviceClient>() }
+                velocity = it.velocity
-
+            ) {
-
+                collectiveModel.locateVisible(it.id).keys
-    LaunchedEffect(collectiveModel) {
-        launchCollectiveMagixServer(collectiveModel)
-
-        withContext(Dispatchers.IO) {
-            val magixClient = MagixEndpoint.rSocketWithWebSockets("localhost")
-
-            client.complete(magixClient)
-
-            collectiveModel.roster.forEach { (id, config) ->
-                scope.launch {
-                    devices[id] = magixClient.remoteDevice(parentContext, "listener", id, id.parseAsName())
-                }
             }
-        }
+            device.moveInCircles()
-
+            it.id to device
-    }
-
-    var selectedDeviceId by remember { mutableStateOf<CollectiveDeviceId?>(null) }
-
-    var currentPosition by remember { mutableStateOf<Gmc?>(null) }
-
-    LaunchedEffect(selectedDeviceId, devices) {
-        selectedDeviceId?.let { devices[it] }?.propertyFlow(CollectiveDevice.position)?.collect {
-            currentPosition = it
         }
     }
 
+    var selectedDeviceId by remember { mutableStateOf<DeviceId?>(null) }
     var showOnlyVisible by remember { mutableStateOf(false) }
 
+    val mapTileProvider = remember {
+        OpenStreetMapTileProvider(
+            client = HttpClient(CIO),
+            cacheDirectory = Path.of("mapCache")
+        )
+    }
+
     HorizontalSplitPane(
         splitPaneState = rememberSplitPaneState(0.9f)
     ) {
         first(400.dp) {
             MapView(
-                mapTileProvider = remember {
+                mapTileProvider = mapTileProvider,
-                    OpenStreetMapTileProvider(
-                        client = HttpClient(CIO),
-                        cacheDirectory = Path.of("mapCache")
-                    )
-                },
                 config = ViewConfig()
             ) {
                 collectiveModel.deviceStates.forEach { device ->
@@ -133,104 +103,82 @@ fun App() {
                     }.launchIn(scope)
                 }
 
-                scope.launch {
+                devices.forEach { (id, device) ->
+                    device.useProperty(CollectiveDevice.position, scope = scope) { position ->
 
-                    client.await().subscribe(DeviceManager.magixFormat).onEach { (magixMessage, deviceMessage) ->
+                        val activeDevice = selectedDeviceId?.let { devices[it] }
-                        if (deviceMessage is PropertyChangedMessage && deviceMessage.property == "position") {
-                            val id = magixMessage.sourceEndpoint
-                            val position = gmcMetaConverter.read(deviceMessage.value)
-                            val activeDevice = selectedDeviceId?.let { devices[it] }
 
-                            if (
+                        if (activeDevice == null || id == selectedDeviceId || !showOnlyVisible || id in activeDevice.listVisible()) {
-                                activeDevice == null ||
+                            rectangle(
-                                id == selectedDeviceId ||
+                                position,
-                                !showOnlyVisible ||
+                                id = id,
-                                id in activeDevice.request(CollectiveDevice.visibleNeighbors)
+                                size = if (selectedDeviceId == id) DpSize(10.dp, 10.dp) else DpSize(5.dp, 5.dp)
-                            ) {
+                            ).color(if (selectedDeviceId == id) Color.Magenta else Color.Blue)
-                                rectangle(
+                                .modifyAttribute(AlphaAttribute, if (selectedDeviceId == id) 1f else 0.5f)
-                                    position,
+                                .onClick { selectedDeviceId = id }
-                                    id = id,
+                        } else {
-                                    size = if (selectedDeviceId == id) DpSize(10.dp, 10.dp) else DpSize(5.dp, 5.dp)
+                            removeFeature(id)
-                                ).color(if (selectedDeviceId == id) Color.Magenta else Color.Blue)
-                                    .modifyAttribute(AlphaAttribute, if (selectedDeviceId == id) 1f else 0.5f)
-                                    .onClick { selectedDeviceId = id }
-                            } else {
-                                removeFeature(id)
-                            }
                         }
-                    }.launchIn(scope)
 
+                    }
                 }
             }
         }
         second(200.dp) {
-
+            Column {
-            Column(
+                selectedDeviceId?.let { id ->
-                modifier = Modifier.verticalScroll(rememberScrollState())
+                    Column(
-            ) {
+                        modifier = Modifier
-                collectiveModel.roster.forEach { (id, _) ->
+                            .padding(8.dp)
-                    Card(
+                            .border(2.dp, Color.DarkGray)
-                        elevation = 16.dp,
-                        modifier = Modifier.padding(8.dp).onClick {
-                            selectedDeviceId = id
-                        }.conditional(id == selectedDeviceId) {
-                            border(2.dp, Color.Blue)
-                        },
                     ) {
-                        Column(
+                        Card(
-                            modifier = Modifier.padding(8.dp)
+                            elevation = 16.dp,
                         ) {
-                            Row(verticalAlignment = Alignment.CenterVertically) {
+                            Text(
-                                if (devices[id] == null) {
+                                text = "Выбран: $id",
-                                    CircularProgressIndicator()
+                                fontSize = 16.sp,
-                                }
+                                fontWeight = FontWeight.Bold,
-                                Text(
+                                modifier = Modifier.padding(10.dp).fillMaxWidth()
-                                    text = id,
+                            )
-                                    fontSize = 16.sp,
+                        }
-                                    fontWeight = FontWeight.Bold,
+                        devices[id]?.let {
-                                    modifier = Modifier.padding(10.dp).fillMaxWidth(),
+                            Text(it.meta.toString(), Modifier.padding(10.dp))
-                                )
+                        }
-                            }
+                        Row(verticalAlignment = Alignment.CenterVertically, modifier = Modifier.padding(10.dp).fillMaxWidth()) {
-                            if (id == selectedDeviceId) {
+                            Text("Показать только видимые")
-                                roster[id]?.let {
+                            Checkbox(showOnlyVisible, { showOnlyVisible = it })
-                                    Text("Meta:", color = Color.Blue, fontWeight = FontWeight.Bold)
+                        }
-                                    Card(elevation = 16.dp, modifier = Modifier.fillMaxWidth().padding(8.dp)) {
+                    }
-                                        Text(it.toString())
+                }
-                                    }
+                Column(
-                                }
+                    modifier = Modifier.verticalScroll(rememberScrollState())
-
+                ) {
-                                currentPosition?.let { currentPosition ->
+                    devices.forEach { (id, device) ->
-                                    Text(
+                        Card(
-                                        "Широта: ${String.format("%.3f", currentPosition.latitude.toDegrees().value)}"
+                            elevation = 16.dp,
-                                    )
+                            modifier = Modifier.padding(8.dp).onClick {
-                                    Text(
+                                selectedDeviceId = id
-                                        "Долгота: ${String.format("%.3f", currentPosition.longitude.toDegrees().value)}"
+                            }.conditional(id == selectedDeviceId) {
-                                    )
+                                border(2.dp, Color.Blue)
-                                    currentPosition.elevation?.let {
-                                        Text("Высота: ${String.format("%.1f", it.meters)} м")
-                                    }
-                                }
-
-                                Row(
-                                    verticalAlignment = Alignment.CenterVertically,
-                                    modifier = Modifier.fillMaxWidth()
-                                ) {
-                                    Text("Показать только видимые")
-                                    Checkbox(showOnlyVisible, { showOnlyVisible = it })
-                                }
                             }
+                        ) {
+                            Text(
+                                text = id,
+                                fontSize = 16.sp,
+                                fontWeight = FontWeight.Bold,
+                                modifier = Modifier.padding(10.dp).fillMaxWidth()
+                            )
                         }
                     }
                 }
             }
-
         }
     }
 }
 
 
 fun main() = application {
-//    System.setProperty(IO_PARALLELISM_PROPERTY_NAME, 300.toString())
     Window(onCloseRequest = ::exitApplication, title = "Maps-kt demo", icon = painterResource("SPC-logo.png")) {
         MaterialTheme {
             App()

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

@@ -83,7 +83,8 @@ suspend fun main() {
 
         val endpointId = "device$it"
         val deviceEndpoint = MagixEndpoint.rSocketStreamWithWebSockets("localhost")
-        deviceManager.launchMagixService(deviceEndpoint, endpointId)
+        deviceManager.launchMagixService(deviceEndpoint, endpointId, Dispatchers.IO)
+
     }
 
     val trace = Bar {

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

@@ -21,10 +21,9 @@ public fun interface MagixFlowPlugin {
         sendMessage: suspend (MagixMessage) -> Unit,
     ): Job
 
+    /**
+     * Use the same [MutableSharedFlow] to send and receive messages. Could be a bottleneck in case of many plugins.
+     */
+    public fun start(scope: CoroutineScope, magixFlow: MutableSharedFlow<MagixMessage>): Job =
+        start(scope, magixFlow) { magixFlow.emit(it) }
 }
-
-/**
- * Use the same [MutableSharedFlow] to send and receive messages. Could be a bottleneck in case of many plugins.
- */
-public fun MagixFlowPlugin.start(scope: CoroutineScope, magixFlow: MutableSharedFlow<MagixMessage>): Job =
-    start(scope, magixFlow) { magixFlow.emit(it) }

magix/magix-server/src/jvmMain/kotlin/space/kscience/magix/server/magixModule.kt

@@ -10,9 +10,7 @@ import io.ktor.server.util.getValue
 import io.ktor.server.websocket.WebSockets
 import io.rsocket.kotlin.ktor.server.RSocketSupport
 import io.rsocket.kotlin.ktor.server.rSocket
-import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.html.*
 import kotlinx.serialization.encodeToString
@@ -44,11 +42,7 @@ private fun ApplicationCall.buildFilter(): MagixMessageFilter {
 /**
  * Attach magix http/sse and websocket-based rsocket event loop + statistics page to existing [MutableSharedFlow]
  */
-public fun Application.magixModule(
+public fun Application.magixModule(magixFlow: MutableSharedFlow<MagixMessage>, route: String = "/") {
-    magixFlow: Flow<MagixMessage>,
-    send: suspend (MagixMessage) -> Unit,
-    route: String = "/",
-) {
     if (pluginOrNull(WebSockets) == null) {
         install(WebSockets)
     }
@@ -68,31 +62,27 @@ public fun Application.magixModule(
 
     routing {
         route(route) {
-            install(ContentNegotiation) {
+            install(ContentNegotiation){
                 json()
             }
-            if (magixFlow is SharedFlow) {
+            get("state") {
-                get("state") {
+                call.respondHtml {
-                    call.respondHtml {
+                    head {
-                        head {
+                        meta {
-                            meta {
+                            httpEquiv = "refresh"
-                                httpEquiv = "refresh"
+                            content = "2"
-                                content = "2"
-                            }
                         }
-                        body {
+                    }
-                            h1 { +"Magix loop statistics" }
+                    body {
-                            if (magixFlow is MutableSharedFlow) {
+                        h1 { +"Magix loop statistics" }
-                                h2 { +"Number of subscribers: ${magixFlow.subscriptionCount.value}" }
+                        h2 { +"Number of subscribers: ${magixFlow.subscriptionCount.value}" }
-                            }
+                        h3 { +"Replay cache size: ${magixFlow.replayCache.size}" }
-                            h3 { +"Replay cache size: ${magixFlow.replayCache.size}" }
+                        h3 { +"Replay cache:" }
-                            h3 { +"Replay cache:" }
+                        ol {
-                            ol {
+                            magixFlow.replayCache.forEach { message ->
-                                magixFlow.replayCache.forEach { message ->
+                                li {
-                                    li {
+                                    code {
-                                        code {
+                                        +magixJson.encodeToString(message)
-                                            +magixJson.encodeToString(message)
-                                        }
                                     }
                                 }
                             }
@@ -112,22 +102,17 @@ public fun Application.magixModule(
             }
             post("broadcast") {
                 val message = call.receive<MagixMessage>()
-                send(message)
+                magixFlow.emit(message)
             }
             //rSocket WS server. Filter from Payload
             rSocket(
                 "rsocket",
-                acceptor = RSocketMagixFlowPlugin.acceptor(application, magixFlow) { send(it) }
+                acceptor = RSocketMagixFlowPlugin.acceptor(application, magixFlow) { magixFlow.emit(it) }
             )
         }
     }
 }
 
-public fun Application.magixModule(
-    magixFlow: MutableSharedFlow<MagixMessage>,
-    route: String = "/",
-): Unit = magixModule(magixFlow, { magixFlow.emit(it) }, route)
-
 /**
  * Create a new loop [MutableSharedFlow] with given [buffer] and setup magix module based on it
  */

magix/magix-server/src/jvmMain/kotlin/space/kscience/magix/server/server.kt

@@ -9,7 +9,6 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import space.kscience.magix.api.MagixEndpoint.Companion.DEFAULT_MAGIX_HTTP_PORT
 import space.kscience.magix.api.MagixFlowPlugin
 import space.kscience.magix.api.MagixMessage
-import space.kscience.magix.api.start
 
 
 /**
@@ -23,6 +22,7 @@ public fun CoroutineScope.startMagixServer(
 
     val magixFlow = MutableSharedFlow<MagixMessage>(
         replay = buffer,
+        extraBufferCapacity = buffer,
         onBufferOverflow = BufferOverflow.DROP_OLDEST
     )