MQTT progress

Author: Robert Winkler

kotlin-wot-binding-http/src/main/kotlin/http/HttpProtocolClient.kt

@@ -26,7 +26,7 @@ import java.util.*
  * Allows consuming Things via HTTP.
  */
 class HttpProtocolClient(
-    private val httpClientConfig: HttpClientConfig?,
+    private val httpClientConfig: HttpClientConfig? = null,
     private val client: HttpClient = HttpClient(CIO)
 ) : ProtocolClient {
 

kotlin-wot-binding-http/src/main/kotlin/http/HttpProtocolServer.kt

@@ -65,7 +65,7 @@ open class HttpProtocolServer(
     }
 
     // Expose a thing
-    override fun expose(thing: ExposedThing) {
+    override suspend fun expose(thing: ExposedThing) {
         if (!started) throw ProtocolServerException("Server has not started yet")
 
         log.info("Exposing thing '{}'", thing.id)
@@ -125,7 +125,7 @@ open class HttpProtocolServer(
                 val observableForm = Form(
                     href = observableHref,
                     contentType = contentType,
-                    op = listOf(Operation.OBSERVE_PROPERTY),
+                    op = listOf(Operation.OBSERVE_PROPERTY, Operation.UNOBSERVE_PROPERTY),
                     subprotocol = "longpoll"
                 )
                 property.forms += observableForm

kotlin-wot-binding-http/src/test/kotlin/http/HttpProtocolServerTest.kt

@@ -374,7 +374,7 @@ class HttpProtocolServerTest {
     }
 
     @Test
-    fun `expose should throw exception if server is not started`() {
+    fun `expose should throw exception if server is not started`() = runTest {
         // Arrange
         server.started = false // Set the server to not started
 
@@ -386,7 +386,7 @@ class HttpProtocolServerTest {
     }
 
     @Test
-    fun `expose should expose thing and add forms`() {
+    fun `expose should expose thing and add forms`() = runTest {
         // Arrange
         server.started = true
         // Act

kotlin-wot-binding-http/src/test/resources/logback.xml

@@ -4,7 +4,7 @@
             <pattern>%d{YYYY-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
         </encoder>
     </appender>
-    <root level="debug">
+    <root level="info">
         <appender-ref ref="STDOUT"/>
     </root>
     <logger name="io.netty" level="INFO"/>

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

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

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

@@ -1,26 +1,23 @@
 package ai.ancf.lmos.wot.binding.mqtt
 
 import ai.ancf.lmos.wot.content.Content
-import ai.ancf.lmos.wot.content.ContentManager
 import ai.ancf.lmos.wot.thing.form.Form
 import ai.anfc.lmos.wot.binding.ProtocolClient
 import ai.anfc.lmos.wot.binding.ProtocolClientException
-import com.hivemq.client.mqtt.MqttGlobalPublishFilter
 import com.hivemq.client.mqtt.datatypes.MqttQos
 import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
 import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
-import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5Subscribe
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.Mqtt5Unsubscribe
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.channelFlow
-import kotlinx.coroutines.flow.onCompletion
+import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.future.await
 import kotlinx.coroutines.suspendCancellableCoroutine
 import org.slf4j.LoggerFactory
 import java.net.URI
 import java.net.URISyntaxException
-import java.net.URL
 import java.util.*
+import java.util.concurrent.ConcurrentHashMap
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 
@@ -33,23 +30,35 @@ class MqttProtocolClient(
     private val log = LoggerFactory.getLogger(MqttProtocolClient::class.java)
     private val scheme = "mqtt" + if (secure) "s" else ""
 
+    private val topicChannels = ConcurrentHashMap<String, Channel<Content>>()
+
     override suspend fun invokeResource(form: Form, content: Content?): Content {
         val topic = try {
             URI(form.href).path.substring(1)
         } catch (e: URISyntaxException) {
             throw ProtocolClientException("Unable to extract topic from href '${form.href}'", e)
         }
-        return requestReply(content, topic)
+        return requestReply(form, content, topic)
     }
 
     override suspend fun subscribeResource(form: Form): Flow<Content> {
         val topic = try {
             URI(form.href).path.substring(1)
-        } catch (e: URISyntaxException) {
+        }
+        catch (e: URISyntaxException) {
             throw ProtocolClientException("Unable to subscribe resource: ${e.message}")
         }
+        return subscribeToTopic(form, topic)
+    }
 
-        return topicObserver(form, topic)
+    override suspend fun unlinkResource(form: Form) {
+        val topic = try {
+            URI(form.href).path.substring(1)
+        }
+        catch (e: URISyntaxException) {
+            throw ProtocolClientException("Unable to unlink resource: ${e.message}")
+        }
+        return unsubscribeFromTopic(topic)
     }
 
     override suspend fun start() {
@@ -60,31 +69,65 @@ class MqttProtocolClient(
         client.disconnect().await()
     }
 
+    // Function to unsubscribe from a topic and close the associated channel
+    private suspend fun unsubscribeFromTopic(topic: String) {
+        // Check if the topic has an associated channel in the ConcurrentHashMap
+        val channel = topicChannels.remove(topic)
+
+        if (channel != null) {
+            try {
+                // Unsubscribe from the topic
+                client.unsubscribeWith()
+                    .topicFilter(topic)
+                    .send()
+                    .await()
+                log.debug("Unsubscribed from topic '{}'", topic)
+            } catch (e: Exception) {
+                log.warn("Error unsubscribing from topic '$topic': ${e.message}")
+            }
+
+            // Close the channel
+            channel.close()
+            log.debug("Closed channel for topic '{}'", topic)
+        } else {
+            log.warn("No active channel found for topic '{}'", topic)
+        }
+    }
+
     // Function to observe a topic using HiveMQ Mqtt5AsyncClient
-    private fun topicObserver(form: Form, topic: String): Flow<Content> = channelFlow {
+    private suspend fun subscribeToTopic(form: Form, topic: String): Flow<Content> {
         log.debug("MqttClient connected to broker at '{}:{}' subscribing to topic '{}'", client.config.serverHost, client.config.serverPort, topic)
 
+        // Create a channel for the topic
+        val channel = Channel<Content>()
+        // Put the channel in the ConcurrentHashMap
+        topicChannels[topic] = channel
+
         try {
             client.subscribeWith()
                 .topicFilter(topic)
-                .qos(MqttQos.AT_LEAST_ONCE)  // QoS level 1
+                .callback() { message ->
+                    log.debug("Received message from topic '{}'", topic)
+                    val content = Content(form.contentType, message.payloadAsBytes)  // Convert payload to Content
+                    val channelResult = channel.trySend(content)
+                    log.debug("Send message to channel flow")
+                }
                 .send()
-                .await()  // Suspending function for subscription completion
+                .await()
 
-            client.publishes(MqttGlobalPublishFilter.SUBSCRIBED) { message ->
-                log.debug("Received message from topic '{}'", topic)
-
-                val content = Content(form.contentType, message.payloadAsBytes)  // Convert payload to Content
-                trySend(content)
-            }
+            log.debug("Subscribed to topic '{}'", topic)
         } catch (e: Exception) {
             log.warn("Error subscribing to topic '$topic': ${e.message}")
-            close(e)  // Close flow on error
+            channel.close(e)
+            //close(e)  // Close flow on error
         }
-    }.onCompletion {
-        val client = client
-        log.debug("No more observers for broker '{}' and topic '{}', unsubscribing.", client.config.serverHost, topic)
+        return channel.consumeAsFlow()
+    }
 
+        /*
+        .onCompletion {
+        val client = client
+        log.debug("No flow collectors anymore, unsubscribing.", client.config.serverHost, topic)
         try {
             client.unsubscribeWith()
                 .topicFilter(topic)
@@ -94,9 +137,11 @@ class MqttProtocolClient(
             log.warn("Error unsubscribing from topic '$topic': ${e.message}")
         }
     }
+    */
+
 
     // Function to publish content to a topic and return a response
-    private suspend fun requestReply(content: Content?, topic: String): Content {
+    private suspend fun requestReply(form: Form, content: Content?, topic: String): Content {
         // Generate a unique response topic for this request
         val responseTopic = "${topic}/reply/${UUID.randomUUID()}"
 
@@ -109,25 +154,26 @@ class MqttProtocolClient(
             )
 
             val payload = content?.body
+            val contentType = form.contentType
 
             // Prepare and send the publish message with a response topic
             val publishMessage = Mqtt5Publish.builder()
                 .topic(topic)
                 .payload(payload)
+                .contentType(contentType)
                 .qos(MqttQos.AT_LEAST_ONCE)
                 .responseTopic(responseTopic)  // Set the response topic
                 .build()
 
             // Publish the message and await reply on the response topic
             return suspendCancellableCoroutine  { continuation ->
-
                 client.subscribeWith()
                     .topicFilter(responseTopic)
                     .qos(MqttQos.AT_LEAST_ONCE)  // QoS level 1 for reliability
                     .callback {
                         response ->
                         log.debug("Response message consumed from topic '$responseTopic'")
-                        val replyContent = content?.type?.let { Content(it, response.payloadAsBytes) } ?: Content.EMPTY_CONTENT
+                        val replyContent = Content(contentType, response.payloadAsBytes)
                         continuation.resume(replyContent)
 
                         // Unsubscribe from the response topic after receiving the response
@@ -163,51 +209,35 @@ class MqttProtocolClient(
         }
     }
 
-    override suspend fun readResource(form: Form): Content = suspendCancellableCoroutine { continuation ->
-        val contentType = form.contentType ?: ContentManager.DEFAULT
-        val requestUri = URL(form.href)
+    // Function to read the resource using the request-reply pattern
+    override suspend fun readResource(form: Form): Content {
+        // Extract the content type from the form or use a default if not provided
 
-        // Extract the topic from the path, removing any leading "/"
-        val filter = requestUri.path.removePrefix("/")
+        // Extract the topic from the URI
+        val requestUri = URI(form.href)
+        val topic = requestUri.path.removePrefix("/") // Removing leading "/"
 
         try {
-
-            Mqtt5Subscribe.builder().topicFilter(filter).build()
-
-            // Subscribing to the topic
-            val subscription = client.subscribe(Mqtt5Subscribe.builder().topicFilter(filter).build())
-            { message ->
-                val content = Content(contentType, message.payloadAsBytes)
-                continuation.resume(content) // Resume the coroutine with the content
-
-                // Unsubscribe after receiving the first message
-                client.unsubscribeWith().topicFilter(filter)
-            }
-
-            // Ensure the subscription is canceled if the coroutine is canceled
-            continuation.invokeOnCancellation {
-                client.unsubscribeWith().topicFilter(filter)
-            }
-
+            // Call requestReply to send a request and get a reply
+            return requestReply(form, null, topic)  // Passing 'null' for content if it's just a read request
         } catch (e: Exception) {
-            // Handle any exception during subscription
-            continuation.resumeWithException(e)
+            // Handle any exception during request-reply
+            throw ProtocolClientException("Failed to read resource from topic '$topic'", e)
         }
     }
 
+    // Function to write the resource using the request-reply pattern
     override suspend fun writeResource(form: Form, content: Content) {
-        val requestUri = URL(form.href)
-        val topic = requestUri.path.removePrefix("/")
-
-        // Publishing message with optional retain and QoS settings
-        val payload = content.body
+        // Extract the topic from the URI
+        val requestUri = URI(form.href)
+        val topic = requestUri.path.removePrefix("/") // Removing leading "/"
 
-        val publishMessage = Mqtt5Publish.builder()
-            .topic(topic)
-            .payload(payload)
-            .qos(MqttQos.AT_LEAST_ONCE)
-            .build()
-
-        client.publish(publishMessage).await()
+        try {
+            // Call requestReply to send the content and get the reply
+            requestReply(form, content, topic)  // Send content to the topic and expect a reply
+        } catch (e: Exception) {
+            // Handle any exception during request-reply
+            throw ProtocolClientException("Failed to write resource to topic '$topic'", e)
+        }
     }
 }

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

@@ -12,7 +12,8 @@ open class MqttProtocolClientFactory(private val mqttClientConfig: MqttClientCon
     override val client: MqttProtocolClient
         get() = MqttProtocolClient(Mqtt5Client.builder()
             .identifier(mqttClientConfig.clientId)
-            .serverHost(mqttClientConfig.broker)
+            .serverHost(mqttClientConfig.host)
+            .serverPort(mqttClientConfig.port)
             .automaticReconnect().applyAutomaticReconnect()
             .build().toAsync())