mirror of
https://github.com/meshtastic/Meshtastic-Android.git
synced 2026-04-20 22:23:37 +00:00
feat(mqtt): adopt mqttastic-client-kmp 0.2.0 — disconnect reasons + Test Connection (#5181)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: jamesarich <2199651+jamesarich@users.noreply.github.com>
This commit is contained in:
parent
5c870028d4
commit
84e70d01a3
12 changed files with 425 additions and 55 deletions
|
|
@ -65,7 +65,6 @@ class MQTTRepositoryImpl(
|
|||
private const val DEFAULT_TOPIC_LEVEL = "/2/e/"
|
||||
private const val JSON_TOPIC_LEVEL = "/2/json/"
|
||||
private const val DEFAULT_SERVER_ADDRESS = "mqtt.meshtastic.org"
|
||||
private const val WEBSOCKET_PATH = "/mqtt"
|
||||
private const val KEEPALIVE_SECONDS = 30
|
||||
private const val INITIAL_RECONNECT_DELAY_MS = 1000L
|
||||
private const val MAX_RECONNECT_DELAY_MS = 30_000L
|
||||
|
|
@ -74,7 +73,7 @@ class MQTTRepositoryImpl(
|
|||
|
||||
@Volatile private var client: MqttClient? = null
|
||||
|
||||
private val _connectionState = MutableStateFlow(ConnectionState.DISCONNECTED)
|
||||
private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected.Idle)
|
||||
override val connectionState: StateFlow<ConnectionState> = _connectionState.asStateFlow()
|
||||
|
||||
@OptIn(ExperimentalSerializationApi::class)
|
||||
|
|
@ -89,7 +88,7 @@ class MQTTRepositoryImpl(
|
|||
Logger.i { "MQTT Disconnecting" }
|
||||
val c = client
|
||||
client = null
|
||||
_connectionState.value = ConnectionState.DISCONNECTED
|
||||
_connectionState.value = ConnectionState.Disconnected.Idle
|
||||
scope.launch { safeCatching { c?.close() }.onFailure { e -> Logger.w(e) { "MQTT clean disconnect failed" } } }
|
||||
}
|
||||
|
||||
|
|
@ -102,14 +101,7 @@ class MQTTRepositoryImpl(
|
|||
val rootTopic = mqttConfig?.root?.ifEmpty { DEFAULT_TOPIC_ROOT } ?: DEFAULT_TOPIC_ROOT
|
||||
|
||||
val rawAddress = mqttConfig?.address ?: DEFAULT_SERVER_ADDRESS
|
||||
val endpoint =
|
||||
if (rawAddress.contains("://")) {
|
||||
MqttEndpoint.parse(rawAddress)
|
||||
} else {
|
||||
// Use WebSocket transport on all platforms for firewall/CDN compatibility.
|
||||
val scheme = if (mqttConfig?.tls_enabled == true) "wss" else "ws"
|
||||
MqttEndpoint.parse("$scheme://$rawAddress$WEBSOCKET_PATH")
|
||||
}
|
||||
val endpoint = resolveEndpoint(rawAddress, mqttConfig?.tls_enabled == true)
|
||||
|
||||
val newClient =
|
||||
MqttClient(ownerId) {
|
||||
|
|
@ -226,3 +218,26 @@ class MQTTRepositoryImpl(
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a user-supplied broker address into an [MqttEndpoint].
|
||||
*
|
||||
* Address resolution rules:
|
||||
* - If [rawAddress] already contains a URI scheme (`scheme://…`), parse it directly via [MqttEndpoint.parse] and
|
||||
* respect whatever transport / port the user encoded.
|
||||
* - Otherwise wrap it as a WebSocket endpoint (`ws[s]://host${WEBSOCKET_PATH}`) so the proxy works over CDNs and
|
||||
* firewall-restricted networks where raw 1883/8883 may be blocked. The scheme is `wss` when [tlsEnabled] is `true`,
|
||||
* `ws` otherwise.
|
||||
*
|
||||
* Extracted as a top-level function so [MQTTRepositoryImplTest] can exercise every branch without spinning up the full
|
||||
* repository, and so `MqttManagerImpl` (in `:core:data`) can reuse the same parsing rules for the probe API. Visibility
|
||||
* is `public` because Kotlin's `internal` is scoped per Gradle module.
|
||||
*/
|
||||
fun resolveEndpoint(rawAddress: String, tlsEnabled: Boolean): MqttEndpoint = if (rawAddress.contains("://")) {
|
||||
MqttEndpoint.parse(rawAddress)
|
||||
} else {
|
||||
val scheme = if (tlsEnabled) "wss" else "ws"
|
||||
MqttEndpoint.parse("$scheme://$rawAddress$WEBSOCKET_PATH")
|
||||
}
|
||||
|
||||
private const val WEBSOCKET_PATH = "/mqtt"
|
||||
|
|
|
|||
|
|
@ -18,25 +18,82 @@ package org.meshtastic.core.network.repository
|
|||
|
||||
import kotlinx.serialization.json.Json
|
||||
import org.meshtastic.core.model.MqttJsonPayload
|
||||
import org.meshtastic.mqtt.MqttEndpoint
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertIs
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class MQTTRepositoryImplTest {
|
||||
|
||||
@Test
|
||||
fun `test address parsing logic`() {
|
||||
val address1 = "mqtt.example.com:1883"
|
||||
val (host1, port1) = address1.split(":", limit = 2).let { it[0] to (it.getOrNull(1)?.toIntOrNull() ?: 1883) }
|
||||
assertEquals("mqtt.example.com", host1)
|
||||
assertEquals(1883, port1)
|
||||
// region resolveEndpoint — every behavioral branch of address parsing.
|
||||
|
||||
val address2 = "mqtt.example.com"
|
||||
val (host2, port2) = address2.split(":", limit = 2).let { it[0] to (it.getOrNull(1)?.toIntOrNull() ?: 1883) }
|
||||
assertEquals("mqtt.example.com", host2)
|
||||
assertEquals(1883, port2)
|
||||
@Test
|
||||
fun `bare host without scheme is wrapped as ws WebSocket on the standard port`() {
|
||||
val endpoint = resolveEndpoint(rawAddress = "broker.example.com", tlsEnabled = false)
|
||||
|
||||
val ws = assertIs<MqttEndpoint.WebSocket>(endpoint)
|
||||
assertEquals("ws://broker.example.com/mqtt", ws.url)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `bare host with TLS enabled is upgraded to wss`() {
|
||||
val endpoint = resolveEndpoint(rawAddress = "broker.example.com", tlsEnabled = true)
|
||||
|
||||
val ws = assertIs<MqttEndpoint.WebSocket>(endpoint)
|
||||
assertEquals("wss://broker.example.com/mqtt", ws.url)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `host with explicit port is preserved when wrapped`() {
|
||||
val endpoint = resolveEndpoint(rawAddress = "broker.example.com:9001", tlsEnabled = false)
|
||||
|
||||
val ws = assertIs<MqttEndpoint.WebSocket>(endpoint)
|
||||
assertEquals("ws://broker.example.com:9001/mqtt", ws.url)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `address with ws scheme is parsed as-is and tls flag is ignored`() {
|
||||
// tlsEnabled is intentionally true here — when the user supplies a full URL we
|
||||
// must honor whatever scheme they provided, not silently upgrade it.
|
||||
val endpoint = resolveEndpoint(rawAddress = "ws://broker.example.com:8080/custom-path", tlsEnabled = true)
|
||||
|
||||
val ws = assertIs<MqttEndpoint.WebSocket>(endpoint)
|
||||
assertEquals("ws://broker.example.com:8080/custom-path", ws.url)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `address with wss scheme is parsed as-is`() {
|
||||
val endpoint = resolveEndpoint(rawAddress = "wss://broker.example.com/secure-mqtt", tlsEnabled = false)
|
||||
|
||||
val ws = assertIs<MqttEndpoint.WebSocket>(endpoint)
|
||||
assertEquals("wss://broker.example.com/secure-mqtt", ws.url)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `address with mqtt tcp scheme is parsed as Tcp endpoint`() {
|
||||
val endpoint = resolveEndpoint(rawAddress = "mqtt://broker.example.com:1883", tlsEnabled = false)
|
||||
|
||||
val tcp = assertIs<MqttEndpoint.Tcp>(endpoint)
|
||||
assertEquals("broker.example.com", tcp.host)
|
||||
assertEquals(1883, tcp.port)
|
||||
assertEquals(false, tcp.tls)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `address with mqtts tcp scheme is parsed as Tcp endpoint with tls true`() {
|
||||
val endpoint = resolveEndpoint(rawAddress = "mqtts://broker.example.com:8883", tlsEnabled = false)
|
||||
|
||||
val tcp = assertIs<MqttEndpoint.Tcp>(endpoint)
|
||||
assertEquals("broker.example.com", tcp.host)
|
||||
assertEquals(8883, tcp.port)
|
||||
assertEquals(true, tcp.tls)
|
||||
}
|
||||
|
||||
// endregion
|
||||
|
||||
// region MqttJsonPayload — keep the existing JSON contract tests.
|
||||
|
||||
@Test
|
||||
fun `test json payload parsing`() {
|
||||
val jsonStr =
|
||||
|
|
@ -72,4 +129,6 @@ class MQTTRepositoryImplTest {
|
|||
assertTrue(jsonStr.contains("\"from\":12345678"))
|
||||
assertTrue(jsonStr.contains("\"payload\":\"Hello World\""))
|
||||
}
|
||||
|
||||
// endregion
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue