MQTT progress

Author: Robert Winkler

kotlin-wot-binding-mqtt/src/main/kotlin/mqtt/MqttClientConfig.kt

@@ -1,9 +1,10 @@
 package ai.ancf.lmos.wot.binding.mqtt
 
+import java.util.*
 
 data class MqttClientConfig(val host: String,
                             val port: Int,
-                            val clientId: String,
+                            val clientId: String = UUID.randomUUID().toString(),
                             private val username: String? = null,
                             private val password: String? = null) {
 }

kotlin-wot-binding-mqtt/src/main/kotlin/mqtt/MqttProtocolClientFactory.kt

@@ -4,18 +4,25 @@ import ai.anfc.lmos.wot.binding.ProtocolClientFactory
 import com.hivemq.client.mqtt.mqtt5.Mqtt5Client
 
 open class MqttProtocolClientFactory(private val mqttClientConfig: MqttClientConfig) : ProtocolClientFactory {
+
+    private val _client: MqttProtocolClient = MqttProtocolClient(
+        Mqtt5Client.builder()
+            .identifier(mqttClientConfig.clientId)
+            .serverHost(mqttClientConfig.host)
+            .serverPort(mqttClientConfig.port)
+            .automaticReconnect()
+            .applyAutomaticReconnect()
+            .build()
+            .toAsync()
+    )
+
     override fun toString(): String {
         return "MqttClient"
     }
     override val scheme: String
         get() = "mqtt"
     override val client: MqttProtocolClient
-        get() = MqttProtocolClient(Mqtt5Client.builder()
-            .identifier(mqttClientConfig.clientId)
-            .serverHost(mqttClientConfig.host)
-            .serverPort(mqttClientConfig.port)
-            .automaticReconnect().applyAutomaticReconnect()
-            .build().toAsync())
+        get() = _client
 
     override suspend fun init() {
         client.start()

kotlin-wot-binding-mqtt/src/main/kotlin/mqtt/MqttProtocolServer.kt

@@ -170,18 +170,30 @@ class MqttProtocolServer(
         }
     }
 
-    private fun exposeTD(thing: ExposedThing) = runBlocking {
+    private suspend fun exposeTD(thing: ExposedThing) {
         val topic = thing.id
         log.debug("Publishing Thing Description '{}' to topic '{}'", thing.id, topic)
 
         try {
-            val content = ContentManager.valueToContent(thing.toJson())
+            val content = ContentManager.valueToContent(thing.toJson(), "application/json")
+            val tdTopic = thing.id
+            client.subscribeWith()
+                .topicFilter(tdTopic)
+                .qos(MqttQos.AT_LEAST_ONCE) // Use AT_LEAST_ONCE QoS level
+                .callback { message ->
+                    CoroutineScope(Dispatchers.IO + exceptionHandler).launch {
+                        respondToTopic(content, message.responseTopic.get())
+                    }
+                }
+                .send().await()  // Sending the subscription request
+            /*
             val publishMessage = Mqtt5Publish.builder()
                 .topic(topic)
                 .payload(content.body)
                 .retain(true)
                 .build()
             client.publish(publishMessage).await()
+            */
         } catch (e: Exception) {
             log.warn("Unable to publish thing description to topic '{}': {}", topic, e.message)
         }

kotlin-wot-binding-mqtt/src/test/kotlin/integration/MqttProtocolServerTest.kt

@@ -62,12 +62,15 @@ class MqttProtocolServerTest {
 
             brokerUrl = "mqtt://$host:${mappedPort}"
 
-            val config = MqttClientConfig(host, mappedPort, "testClient")
+            val clientConfig = MqttClientConfig(host, mappedPort, "client")
+            val serverConfig = MqttClientConfig(host, mappedPort, "server")
+
+            val clientFactory = MqttProtocolClientFactory(clientConfig)
+            clientFactory.init()
+            protocolClient = clientFactory.client
 
-            protocolClient = MqttProtocolClientFactory(config).client
-            protocolClient.start()
             val servient = Servient()
-            mqttServer = MqttProtocolServer(config)
+            mqttServer = MqttProtocolServer(serverConfig)
             mqttServer.start(servient)
             exposedThing = exposedThing(servient, id = "test") {
                 stringProperty(PROPERTY_NAME) {

kotlin-wot-integration-tests/src/test/kotlin/integration/WoTMqttIntegrationTest.kt

@@ -57,12 +57,12 @@ class WoTMqttIntegrationTest() {
 
     @Test
     fun `Should fetch thing`() = runTest {
-        val config = MqttClientConfig(hiveMqContainer.host,
-            hiveMqContainer.getMappedPort(1883), "testClient")
+        val clientConfig = MqttClientConfig(hiveMqContainer.host, hiveMqContainer.getMappedPort(1883), "client")
+        val serverConfig = MqttClientConfig(hiveMqContainer.host, hiveMqContainer.getMappedPort(1883), "server")
 
         val servient = Servient(
-            servers = listOf(MqttProtocolServer(config)),
-            clientFactories = listOf(MqttProtocolClientFactory(config))
+            servers = listOf(MqttProtocolServer(serverConfig)),
+            clientFactories = listOf(MqttProtocolClientFactory(clientConfig))
         )
         val wot = Wot.create(servient)