Implement MQTT Message Publishing in MqttManager (CATROID-1673)

catroid/src/main/java/org/catrobat/catroid/devices/mqtt/MqttClientInterface.kt

@@ -32,4 +32,5 @@ interface MqttClientInterface {
     fun disconnect()
     fun close()
     fun setCallback(callback: MqttCallback)
+    fun publish(topic: String, payload: ByteArray, qos: Int, retained: Boolean)
 }

catroid/src/main/java/org/catrobat/catroid/devices/mqtt/MqttManager.kt

@@ -41,6 +41,7 @@ class MqttManager(private val clientFactory: MqttClientFactory = DefaultMqttClie
         get() = mqttClient?.isConnected == true
 
     companion object {
+        private const val NULL_CHARACTER = '\u0000'
         private val TAG = MqttManager::class.simpleName
         private const val TCP_SCHEME = "tcp"
         private const val SSL_SCHEME = "ssl"
@@ -88,6 +89,47 @@ class MqttManager(private val clientFactory: MqttClientFactory = DefaultMqttClie
         }
     }
 
+    fun publishFromContext(context: Context, topic: String, payload: String, qos: Int = 0, retained: Boolean = false) =
+        publish(MqttConnectionConfig.fromContext(context), topic, payload, qos, retained)
+
+    fun publish(config: MqttConnectionConfig, topic: String, payload: String, qos: Int = 0, retained: Boolean = false): Boolean {
+        if (topic.isBlank()) {
+            Log.e(TAG, "Cannot publish: topic is blank")
+            return false
+        }
+        if (topic.contains('#') || topic.contains('+')) {
+            Log.e(TAG, "Cannot publish: topic contains wildcard characters")
+            return false
+        }
+        if (topic.contains(NULL_CHARACTER)) {
+            Log.e(TAG, "Cannot publish: topic contains a null character")
+            return false
+        }
+        if (qos !in 0..2) {
+            Log.e(TAG, "Cannot publish: invalid QoS value $qos")
+            return false
+        }
+        if (!isConnected && !connect(config)) {
+            Log.e(TAG, "Cannot publish: connection failed")
+            return false
+        }
+        val client = mqttClient ?: run {
+            Log.e(TAG, "Cannot publish: no client available")
+            return false
+        }
+        return try {
+            client.publish(topic, payload.toByteArray(), qos, retained)
+            true
+        } catch (e: MqttException) {
+            Log.e(TAG, "Failed to publish to '$topic'", e)
+            false
+        } catch (e: IllegalArgumentException) {
+            // Paho validates the topic itself and throws this, e.g. when it exceeds 65535 bytes.
+            Log.e(TAG, "Broker rejected the topic '$topic'", e)
+            false
+        }
+    }
+
     fun disconnect() {
         synchronized(this) {
             if (mqttClient == null) return
@@ -119,11 +161,9 @@ class MqttManager(private val clientFactory: MqttClientFactory = DefaultMqttClie
 
     private val callback = object : MqttCallback {
         override fun connectionLost(cause: Throwable?) {
-            Log.e(TAG, "Connection lost: ${cause?.message}")
+            Log.e(TAG, "Connection lost", cause)
         }
-        // Message handling is implemented in a later ticket.
         override fun messageArrived(topic: String, message: MqttMessage) = Unit
-        // Delivery tokens are not used until publish is implemented in a later ticket.
         override fun deliveryComplete(token: IMqttDeliveryToken?) = Unit
     }
 }

catroid/src/main/java/org/catrobat/catroid/devices/mqtt/PahoMqttClient.kt

@@ -26,6 +26,7 @@ package org.catrobat.catroid.devices.mqtt
 import org.eclipse.paho.client.mqttv3.MqttCallback
 import org.eclipse.paho.client.mqttv3.MqttClient
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions
+import org.eclipse.paho.client.mqttv3.MqttMessage
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
 
 class PahoMqttClient(brokerUrl: String, clientId: String) : MqttClientInterface {
@@ -35,4 +36,11 @@ class PahoMqttClient(brokerUrl: String, clientId: String) : MqttClientInterface
     override fun disconnect() = client.disconnect()
     override fun close() = client.close()
     override fun setCallback(callback: MqttCallback) = client.setCallback(callback)
+    override fun publish(topic: String, payload: ByteArray, qos: Int, retained: Boolean) {
+        val message = MqttMessage(payload).apply {
+            this.qos = qos
+            isRetained = retained
+        }
+        client.publish(topic, message)
+    }
 }

catroid/src/test/java/org/catrobat/catroid/test/mqtt/MqttManagerTest.kt

@@ -256,6 +256,162 @@ class MqttManagerTest {
         // no exception = pass
     }
 
+    // --- publish() ---
+
+    @Test
+    fun testPublishReturnsTrueWhenConnectedAndTopicValid() {
+        manager.connect(defaultConfig)
+        assertTrue(manager.publish(defaultConfig, "home/temp", "22"))
+    }
+
+    @Test
+    fun testPublishCallsClientPublish() {
+        manager.connect(defaultConfig)
+        manager.publish(defaultConfig, "home/temp", "22")
+        assertTrue(fakeClient.publishCalled)
+    }
+
+    @Test
+    fun testPublishSendsCorrectTopicAndPayload() {
+        manager.connect(defaultConfig)
+        manager.publish(defaultConfig, "home/temp", "42")
+        assertEquals("home/temp", fakeClient.lastTopic)
+        assertEquals("42", fakeClient.lastPayload)
+    }
+
+    @Test
+    fun testPublishSetsQosAndRetained() {
+        manager.connect(defaultConfig)
+        manager.publish(defaultConfig, "home/temp", "on", qos = 2, retained = true)
+        assertEquals(2, fakeClient.lastQos)
+        assertTrue(fakeClient.lastRetained)
+    }
+
+    @Test
+    fun testPublishTriggersLazyConnectWhenDisconnected() {
+        fakeClient.connected = false
+        manager.publish(defaultConfig, "home/temp", "22")
+        assertTrue(fakeClient.connectCalled)
+    }
+
+    @Test
+    fun testPublishReturnsTrueAfterLazyConnect() {
+        fakeClient.connected = false
+        assertTrue(manager.publish(defaultConfig, "home/temp", "22"))
+    }
+
+    @Test
+    fun testPublishReturnsFalseWhenLazyConnectFails() {
+        fakeClient.connected = false
+        fakeClient.throwOnConnect = true
+        assertFalse(manager.publish(defaultConfig, "home/temp", "22"))
+    }
+
+    @Test
+    fun testPublishReturnsFalseForBlankTopic() {
+        manager.connect(defaultConfig)
+        assertFalse(manager.publish(defaultConfig, "   ", "22"))
+    }
+
+    @Test
+    fun testPublishDoesNotCallClientForBlankTopic() {
+        manager.connect(defaultConfig)
+        manager.publish(defaultConfig, "   ", "22")
+        assertFalse(fakeClient.publishCalled)
+    }
+
+    @Test
+    fun testPublishReturnsFalseForTopicWithHashWildcard() {
+        manager.connect(defaultConfig)
+        assertFalse(manager.publish(defaultConfig, "home/#", "22"))
+    }
+
+    @Test
+    fun testPublishReturnsFalseForTopicWithPlusWildcard() {
+        manager.connect(defaultConfig)
+        assertFalse(manager.publish(defaultConfig, "home/+/temp", "22"))
+    }
+
+    @Test
+    fun testPublishReturnsFalseForTopicWithNullCharacter() {
+        manager.connect(defaultConfig)
+        assertFalse(manager.publish(defaultConfig, "home/" + 0.toChar() + "temp", "22"))
+    }
+
+    @Test
+    fun testPublishDoesNotCallClientForTopicWithNullCharacter() {
+        manager.connect(defaultConfig)
+        manager.publish(defaultConfig, "home/" + 0.toChar() + "temp", "22")
+        assertFalse(fakeClient.publishCalled)
+    }
+
+    @Test
+    fun testPublishReturnsFalseWhenClientThrowsIllegalArgument() {
+        manager.connect(defaultConfig)
+        fakeClient.throwIllegalArgumentOnPublish = true
+        assertFalse(manager.publish(defaultConfig, "home/temp", "22"))
+    }
+
+    @Test
+    fun testPublishReturnsFalseForInvalidQos() {
+        manager.connect(defaultConfig)
+        assertFalse(manager.publish(defaultConfig, "home/temp", "22", qos = 3))
+    }
+
+    @Test
+    fun testPublishReturnsFalseWhenClientThrows() {
+        manager.connect(defaultConfig)
+        fakeClient.throwOnPublish = true
+        assertFalse(manager.publish(defaultConfig, "home/temp", "22"))
+    }
+
+    @Test
+    fun testPublishWithEmptyPayloadReturnsTrue() {
+        manager.connect(defaultConfig)
+        assertTrue(manager.publish(defaultConfig, "home/temp", ""))
+    }
+
+    @Test
+    fun testPublishWithQosZeroReturnsTrue() {
+        manager.connect(defaultConfig)
+        assertTrue(manager.publish(defaultConfig, "home/temp", "22", qos = 0))
+    }
+
+    @Test
+    fun testPublishWithQosOneReturnsTrue() {
+        manager.connect(defaultConfig)
+        assertTrue(manager.publish(defaultConfig, "home/temp", "22", qos = 1))
+    }
+
+    @Test
+    fun testPublishWithQosTwoReturnsTrue() {
+        manager.connect(defaultConfig)
+        assertTrue(manager.publish(defaultConfig, "home/temp", "22", qos = 2))
+    }
+
+    @Test
+    fun testPublishWithRetainedFalseSetsRetainedFalse() {
+        manager.connect(defaultConfig)
+        manager.publish(defaultConfig, "home/temp", "22", retained = false)
+        assertFalse(fakeClient.lastRetained)
+    }
+
+    @Test
+    fun testPublishWhenAlreadyConnectedDoesNotReconnect() {
+        manager.connect(defaultConfig)
+        fakeClient.connectCalled = false
+        manager.publish(defaultConfig, "home/temp", "22")
+        assertFalse(fakeClient.connectCalled)
+    }
+
+    @Test
+    fun testPublishDoesNotCallClientWhenLazyConnectFails() {
+        fakeClient.connected = false
+        fakeClient.throwOnConnect = true
+        manager.publish(defaultConfig, "home/temp", "22")
+        assertFalse(fakeClient.publishCalled)
+    }
+
     // --- FakeMqttClientFactory ---
 
     private class FakeMqttClientFactory(private val client: FakeMqttClient) : MqttClientFactory {
@@ -275,6 +431,13 @@ class MqttManagerTest {
         var closeCalled = false
         var callbackSet = false
         var throwOnConnect = false
+        var throwOnPublish = false
+        var throwIllegalArgumentOnPublish = false
+        var publishCalled = false
+        var lastTopic: String? = null
+        var lastPayload: String? = null
+        var lastQos: Int = -1
+        var lastRetained: Boolean = false
         var lastConnectOptions: MqttConnectOptions? = null
 
         override val isConnected get() = connected
@@ -298,5 +461,15 @@ class MqttManagerTest {
         override fun setCallback(callback: MqttCallback) {
             callbackSet = true
         }
+
+        override fun publish(topic: String, payload: ByteArray, qos: Int, retained: Boolean) {
+            if (throwOnPublish) throw org.eclipse.paho.client.mqttv3.MqttException(0)
+            if (throwIllegalArgumentOnPublish) throw IllegalArgumentException("Invalid topic")
+            publishCalled = true
+            lastTopic = topic
+            lastPayload = String(payload)
+            lastQos = qos
+            lastRetained = retained
+        }
     }
 }