MqttBox.kt
src/jvmMain/kotlin/afield/io/mqtt/MqttBox.kt · 354 lines
@file:afield.runtime.completion.CompletionCategory("Networking")
package afield.io.mqtt
import afield.document.graph.Box
import afield.runtime.lifecycle.Runloop
import kotlinx.serialization.json.JsonArray
import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonNull
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonArray
import kotlinx.serialization.json.buildJsonObject
import java.net.BindException
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueueThe box-facing MQTT surface: mqtt and mqtt(url) return a BoxMqtt for publishing and
subscribing, and mqttBroker(port) starts an embedded broker.
Callbacks arrive on the main Runloop, so they can touch boxes, sliders and graphics directly. Connections and brokers outlive a stopped box and are shared by url and port; subscriptions belong to the box that made them. MqttConnections holds that shared state.
object MqttConnections
Process-wide MQTT state behind Box.mqtt: one client per broker url, the embedded brokers, and
the queue that brings callbacks onto the main Runloop.
Lifecycle follows the house rule for launched things: clients and brokers are keyed (by url and
by port) and survive a box being stopped — a sensor feed should not drop because you stopped
the box that plots it. Re-running a box reuses them. Subscriptions are keyed by box and key, so a
re-run replaces its callback instead of adding a second one. They are owned by the box as
ReleasePolicy.KEYED, so deleting the box (or closing its document) unsubscribes them; otherwise:
BoxMqtt.unsubscribe, BoxMqtt.close, or closeAll from a "panic" box.
object MqttConnections {
private val clients = ConcurrentHashMap<String, MqttClient>()
private val brokers = ConcurrentHashMap<Int, MqttBroker>()
private val slots = ConcurrentHashMap<String, Slot>()
private val histories = ConcurrentHashMap<String, ConcurrentHashMap<String, History>>()var historyLength
How many numeric values per topic BoxMqtt.series keeps.
@Volatile
var historyLength = 512
private val queue = ConcurrentLinkedQueue<Runnable>()
@Volatile private var drainInstalled = falsefun dispatch(r: Runnable)
Runs r on the main Runloop when there is one (inside Field), directly otherwise (tests,
jvm-main, where no Runloop turns and a queued callback would never run).
fun dispatch(r: Runnable) {
if (Runloop.mainThread == null) return r.run()
queue.add(r)
if (!drainInstalled) synchronized(this) {
if (!drainInstalled) {
drainInstalled = true
Runloop.runLoop.preRun.setUnowned("afield.mqtt.dispatch", sequence {
while (true) {
// Bounded per frame, so a flood of messages cannot stall the Runloop.
var n = 0
while (n++ < 2000) {
val next = queue.poll() ?: break
try { next.run() } catch (t: Throwable) { afield.app.diagnostics.logging.lprintln("error,mqtt", "callback threw: $t") }
}
yield(null)
}
})
}
}
}internal class Slot(val filter: String, val client: MqttClient, @Volatile var handler: (String, MqttMessage) -> Unit) {
lateinit var subscription: MqttSubscription
internal class History(capacity: Int)
internal class History(capacity: Int) {
private val values = DoubleArray(capacity)
private var start = 0
private var size = 0@Synchronized fun add(v: Double)
@Synchronized fun add(v: Double) {
if (size < values.size) values[(start + size++) % values.size] = v
else { values[start] = v; start = (start + 1) % values.size }
}@Synchronized fun toList(): List<Double> = List(size) { values[(start + it) % values.size] }
fun broker(port: Int = Mqtt.PORT, bindAddress: String = "127.0.0.1"): MqttBroker
The embedded broker on port, started on first use and kept until closeAll.
fun broker(port: Int = Mqtt.PORT, bindAddress: String = "127.0.0.1"): MqttBroker =
brokers.compute(port) { _, old -> if (old != null && !old.isClosed) old else MqttBroker(port, bindAddress) }!!val defaultUrl: String by lazy
The url Box.mqtt connects to: AFIELD_MQTT_URL if set, else an embedded broker on 1883, or
on a free port if something else already has 1883.
val defaultUrl: String by lazy {
System.getenv("AFIELD_MQTT_URL")?.takeIf { it.isNotBlank() } ?: try {
broker(Mqtt.PORT).url
} catch (_: BindException) {
broker(0).also { brokers[it.port] = it }.url
}
}fun client(url: String, clientId: String? = null, will: MqttPacket.Will? = null, username: String? = null, password: String? = null): MqttClient
The client for url, made (and connected in the background) on first use.
fun client(url: String, clientId: String? = null, will: MqttPacket.Will? = null, username: String? = null, password: String? = null): MqttClient {
var created = false
return clients.compute(url) { _, old ->
if (old != null && !old.isClosed) old
else MqttClient(
url, clientId = clientId ?: ("field4-" + java.util.UUID.randomUUID().toString().take(8)),
will = will, username = username, password = password, dispatch = ::dispatch,
).also { c -> created = true; watch(url, c) }
}!!.also { c ->
// A box's first publish usually follows straight on; give a fresh connection a moment so
// QoS 0 messages are not dropped while it connects. An unreachable broker costs this once.
if (created) c.awaitConnected(1000)
}
}
private fun watch(url: String, c: MqttClient) {
val h = histories.computeIfAbsent(url) { ConcurrentHashMap() }
// Recorded on the reader thread from whatever the client's subscriptions bring in.
fun add(key: String, v: Double) = h.computeIfAbsent(key) { History(historyLength) }.add(v)
c.onReceive.add { m ->
val plain = m.text.trim().toDoubleOrNull()
if (plain != null) add(m.topic, plain)
else if (m.payload.size < 4096 && m.payload.firstOrNull()?.toInt()?.toChar() in setOf('{', '[')) {
// JSON: every numeric leaf, under "topic|dotted.path", for series(topic, path).
fun walk(e: JsonElement, path: String, depth: Int) {
when (e) {
is JsonPrimitive -> if (!e.isString) e.content.toDoubleOrNull()?.let { add(if (path.isEmpty()) m.topic else "${m.topic}|$path", it) }
is JsonObject -> if (depth < 4) e.forEach { (k, v) -> walk(v, if (path.isEmpty()) k else "$path.$k", depth + 1) }
is JsonArray -> if (depth < 4 && e.size <= 64) e.forEachIndexed { i, v -> walk(v, if (path.isEmpty()) "$i" else "$path.$i", depth + 1) }
}
}
m.jsonOrNull?.let { walk(it, "", 0) }
}
}
}internal fun history(url: String, topic: String, path: String? = null): List<Double>
internal fun history(url: String, topic: String, path: String? = null): List<Double> =
histories[url]?.get(if (path == null) topic else "$topic|$path")?.toList() ?: emptyList()internal fun subscribe(box: Box, client: MqttClient, key: String, filter: String, qos: QoS, handler: (String, MqttMessage) -> Unit) {
internal fun subscribe(box: Box, client: MqttClient, key: String, filter: String, qos: QoS, handler: (String, MqttMessage) -> Unit) {
val k = "${box.id}|${client.url}|$key"
slots.compute(k) { _, old ->
if (old != null && old.filter == filter && old.client === client && !old.subscription.isClosed) {
old.handler = handler; old
} else {
old?.subscription?.close()
Slot(filter, client, handler).also { slot ->
// KEYED: stop leaves it subscribed; deleting the box (or closing the document) releases it.
box.runScopes.parent.own(afield.runtime.lifecycle.ReleasePolicy.KEYED) {
if (slots.remove(k, slot)) slot.subscription.close()
}
slot.subscription = client.subscribe(filter, qos) { m ->
Box.withBoxContextNow(box) {
try { slot.handler(m.topic, m) } catch (t: Throwable) { box.println("mqtt '$filter': $t") }
}
}
}
}
}
}internal fun unsubscribe(box: Box, url: String?, key: String?)
internal fun unsubscribe(box: Box, url: String?, key: String?) {
slots.entries.removeIf { (k, slot) ->
val parts = k.split('|', limit = 3)
val hit = parts[0] == box.id && (url == null || parts[1] == url) && (key == null || parts[2] == key)
if (hit) slot.subscription.close()
hit
}
}internal fun subscriptionKeys(box: Box): List<String>
internal fun subscriptionKeys(box: Box): List<String> = slots.keys.filter { it.startsWith("${box.id}|") }.map { it.split('|', limit = 3)[2] }fun closeClient(url: String)
fun closeClient(url: String) {
clients.remove(url)?.close()
slots.entries.removeIf { it.value.client.url == url }
}fun closeAll()
Close every client and embedded broker.
fun closeAll() {
slots.clear()
clients.values.forEach { it.close() }; clients.clear()
brokers.values.forEach { it.close() }; brokers.clear()
}
}class BoxMqtt internal constructor(private val box: Box, val client: MqttClient)
MQTT from a box, against one broker. Get one with mqtt (the default broker) or
mqtt("tcp://192.168.1.10:1883").
class BoxMqtt internal constructor(private val box: Box, val client: MqttClient) {val url: String get() = client.url
val connected: Boolean get() = client.isConnected
fun subscribe(filter: String, qos: Int = 0, key: String = filter, handler: (topic: String, message: MqttMessage) -> Unit) {
Call handler with (topic, message) for every message matching filter. + matches one
level (sensors/+/temp), # everything below (sensors/#). Runs on the main Runloop.
Re-running the box replaces the handler for the same key (the filter by default), so a box
can be edited and re-run without callbacks piling up. Stopping the box leaves it subscribed;
unsubscribe ends it.
fun subscribe(filter: String, qos: Int = 0, key: String = filter, handler: (topic: String, message: MqttMessage) -> Unit) {
MqttConnections.subscribe(box, client, key, filter, QoS.of(qos), handler)
}fun track(filter: String, qos: Int = 0)
Subscribe to filter with no callback, so latest, number and series fill in for its
topics. A connection only receives topics something has subscribed to.
fun track(filter: String, qos: Int = 0) = subscribe(filter, qos, key = "track:$filter") { _, _ -> }fun toOsc(filter: String, osc: afield.io.osc.OscClient, address: (String) -> String
Forward every message matching filter as OSC through osc (oscClient(host, port)), to the
address address makes from the topic (by default the topic itself: sensors/a/temp goes to
/sensors/a/temp). A numeric payload is sent as one float, numeric JSON fields as one message
each at <address>/<field>, anything else as one string.
fun toOsc(filter: String, osc: afield.io.osc.OscClient, address: (String) -> String = { "/$it" }) =
subscribe(filter, key = "osc:$filter") { topic, m ->
val base = address(topic)
val n = m.text.trim().toDoubleOrNull()
val json = if (n == null) m.jsonOrNull as? JsonObject else null
when {
n != null -> osc.send(base, n.toFloat())
json != null -> json.forEach { (k, v) ->
(v as? JsonPrimitive)?.takeIf { !it.isString }?.content?.toDoubleOrNull()?.let { osc.send("$base/$k", it.toFloat()) }
}
else -> osc.send(base, m.text)
}
}fun toDmx
Drive one DMX channel from a numeric topic: values in range map to 0..255 on 1-based
channel of universe in out (artnetOut(...) / sacnOut(...)). path reads a JSON field.
fun toDmx(
topic: String,
out: afield.io.dmx.DmxOutput,
universe: Int,
channel: Int,
path: String? = null,
range: ClosedFloatingPointRange<Double> = 0.0..1.0,
) = subscribe(topic, key = "dmx:$topic|$path|$universe:$channel") { _, m ->
val v = (if (path == null) m.number else m.number(path)) ?: return@subscribe
out.universe(universe).setLevel(channel, (v - range.start) / (range.endInclusive - range.start))
}fun unsubscribe(key: String? = null) = MqttConnections.unsubscribe(box, client.url, key)
End this box's subscription under key, or all of them on this broker.
val subscriptions: List<String> get() = MqttConnections.subscriptionKeys(box)
This box's subscription keys on this broker.
fun publish(topic: String, value: Any?, qos: Int = 0, retain: Boolean = false)
Publish value to topic. A String or number goes as text, a ByteArray as is, and a Map,
List or JsonElement as JSON. retain keeps it on the broker for later subscribers.
fun publish(topic: String, value: Any?, qos: Int = 0, retain: Boolean = false) {
client.publish(topic, encode(value), QoS.of(qos), retain)
}fun clearRetained(topic: String)
Clear a retained message.
fun clearRetained(topic: String) {
client.publish(topic, ByteArray(0), QoS.AT_MOST_ONCE, retain = true)
}val latest: Map<String, MqttMessage> get() = client.latest
The latest message on every topic this connection has received.
operator fun get(topic: String): MqttMessage? = client.latest[topic]
The latest message on topic, or null.
fun number(topic: String, path: String? = null, default: Double = 0.0): Double
The latest numeric value on topic (or default); with path, a number inside JSON.
fun number(topic: String, path: String? = null, default: Double = 0.0): Double {
val m = client.latest[topic] ?: return default
return (if (path == null) m.number else m.number(path)) ?: default
}fun text(topic: String, default: String = ""): String = client.latest[topic]?.text ?: default
The latest text on topic, or default.
fun series(topic: String, path: String? = null): List<Double>
The recent numeric values on topic, oldest first (up to MqttConnections.historyLength), for
showSignals("temp" to mqtt.series("sensors/a/temp")). With path, a number inside a JSON
payload, as in MqttMessage.number: series("sensors/garden/env", "temp").
fun series(topic: String, path: String? = null): List<Double> = MqttConnections.history(client.url, topic, path)fun close() = MqttConnections.closeClient(client.url)
Disconnect from this broker and forget its subscriptions (every box's).
override fun toString()
override fun toString() = "mqtt ${client.url}: ${if (connected) "connected" else "connecting (${client.lastError ?: "…"})"}, " +
"${client.received.get()} received, ${client.sent.get()} sent, ${client.latest.size} topics"companion object
companion object {fun encode(value: Any?): ByteArray
How publish turns a value into bytes.
fun encode(value: Any?): ByteArray = when (value) {
null -> ByteArray(0)
is ByteArray -> value
is String -> value.encodeToByteArray()
is Number, is Boolean -> value.toString().encodeToByteArray()
else -> toJson(value).toString().encodeToByteArray()
}fun toJson(value: Any?): JsonElement
Maps, lists, numbers, strings and booleans to JSON.
fun toJson(value: Any?): JsonElement = when (value) {
null -> JsonNull
is JsonElement -> value
is Number -> JsonPrimitive(value)
is Boolean -> JsonPrimitive(value)
is String -> JsonPrimitive(value)
is Map<*, *> -> buildJsonObject { value.forEach { (k, v) -> put(k.toString(), toJson(v)) } }
is Iterable<*> -> buildJsonArray { value.forEach { add(toJson(it)) } }
is Array<*> -> toJson(value.toList())
is DoubleArray -> toJson(value.toList())
is FloatArray -> toJson(value.toList())
is IntArray -> toJson(value.toList())
else -> JsonPrimitive(value.toString())
}
}
}val Box.mqtt: BoxMqtt get() = BoxMqtt(this, MqttConnections.client(MqttConnections.defaultUrl))
MQTT against the default broker (an embedded one unless AFIELD_MQTT_URL is set).
fun Box.mqtt
MQTT against the broker at url (tcp://host:1883). The first call for a url makes the
connection, with clientId, will (topic to payload, published retained by the broker if Field
vanishes) and credentials; later calls reuse it.
fun Box.mqtt(
url: String,
clientId: String? = null,
will: Pair<String, String>? = null,
username: String? = null,
password: String? = null,
): BoxMqtt = BoxMqtt(this, MqttConnections.client(
url, clientId, will?.let { MqttPacket.Will(it.first, it.second.encodeToByteArray(), QoS.AT_LEAST_ONCE, retain = true) }, username, password,
))fun Box.mqttBroker(port: Int = Mqtt.PORT, bindAddress: String = "127.0.0.1"): MqttBroker
Start (or reuse) an embedded broker on port. Bind to "0.0.0.0" so devices on the network can
connect to this machine. It keeps running when the box stops.
fun Box.mqttBroker(port: Int = Mqtt.PORT, bindAddress: String = "127.0.0.1"): MqttBroker = MqttConnections.broker(port, bindAddress)