Slow Initial Data Transmission Issue in MQTT Client

[SnowCharmQ]

🐛 Bug Report

I encapsulate an MQTT service client. This API will monitor the input data and publish the data to the MQTT broker when the data changes. However, when I computed the throughput of the service, I noticed that the transmission speed of the first batch samples of data was extremely slow, taking 10 seconds or even longer. The transmission speed of the subsequent data sent meets expectations. Based on the calculated time, the printed results are as follows: (on the receiver side, but the result is the same on the sender side, there is much useless information in Logcat)

Is it a warm-up process? Or just a bug? This bug could lead to a loss of real-time performance in the program.

Code sample

    @OptIn(DelicateCoroutinesApi::class)
    fun request(
        qos: QosOption,
        data: MutableStateFlow<Pair<Long, DoubleArray>>
    ) {
        val mqttQos = qos.toQos()
        thread {
            val client: Mqtt5AsyncClient = MqttClient.builder()
                .useMqttVersion5()
                .identifier(mqttTask.identifier)
                .serverHost(GlobalConfig.mqtt.mqttIp)
                .serverPort(GlobalConfig.mqtt.mqttPort)
                .buildAsync()
            client.connect()
                .whenComplete { _: Mqtt5ConnAck?, throwable: Throwable? ->
                    if (throwable != null) {
                        return@whenComplete
                    } else {
                        var s = System.currentTimeMillis()
                        var cnt = 0
                        GlobalScope.launch(Dispatchers.IO) {
                            data.collect {
                                if (it.first == 0L) return@collect
                                client.publishWith()
                                    .topic(mqttTask.topic)
                                    .qos(mqttQos)
                                    .payload(it.toString().toByteArray())
                                    .send()
                                cnt += 1
                                if (cnt % GlobalConfig.sampleRate == 0) {
                                    val e = System.currentTimeMillis()
                                    println("${e - s} $cnt")
                                    s = e
                                }
                            }
                        }
                    }
                }
        }
    }

Environment

Windows 10

Android sdk33

Version: MQTT v5.0

MQTT Broker: EMQX

📈 Expected behavior

All batches of data should be transferred in 1s.

[SnowCharmQ]

I can ensure that the quantity of data is large enough to be transmitted.

[pglombardo]

Hi @SnowCharmQ,

My initial thought would be at least partially due to warm up but >10 seconds seems extreme. I also see you connect on every call to request.

A couple ideas to potentially reduce the problem area would be:

1. To rule out any broker issues, could you try with another? Maybe test messages to a any non EMQX public broker?
2. You could keep a pre-initialized and connected client that is reused on each call to request. This should avoid repeating the client initialization and TCP connection establishment.
3. You could profile the broker connection - how long is client.connect() on the first call versus subsequent?

Visibility is likely key here - The results from one of the above should put us on the right path...