Mqtt3Client Flowable Timeout Exception

Hello everyone

Im working with HiveMq Client Mqtt3 for Android and when i called Flowable Subscribe function i receive a TimeOutException

I’ve tried to set timeOut property but without success.

I called connect().andThen(subscribeTopic().ignoreElements())

1 - Connect OK
2 - Subscribe OK but the subscription returned in onError() -> The source did not signal an event for 6 seconds and has been terminated.

Does anyone know how can avoid this exception? Because this stop the Thread and don’t execute another .andThen in sequence… Example: connect().andThen(subscribe()).andThen(disconnect())

Virgilio Magalhaes

Hi Virgilio,

Nice to see your interest in MQTT and the HiveMQ MQTT client.
Can you please share a more complete snippet of your code, as well as the error you are seeing?

Thank you and regards,

Hello Florian,

follow part of my code. Thank you.

                .doOnSubscribe { Timber.d("App onSubscribe") }
                .timeout(6L, TimeUnit.SECONDS)
                    { Timber.d("App onComplete") },
                    { Timber.d("App onError ${it.message}") }
                ).apply { disposable.add(this) }

Provider setup will call init function in my Provider, the function always returned for callback onError with Exception message described before

init {

fun connect(): Completable {

        if (this.client.state == MqttClientState.CONNECTED ||
            this.client.state == MqttClientState.CONNECTING) {
            Timber.i("MQTT BROKER >>> MqttClientState.CONNECTED")
            return Completable.complete()

        return this.client.connect(mqttConnect)
            .doOnSuccess {
                Timber.i("MQTT BROKER >>> CONNECTED WITH HIVEMQ CLIENT::: ${it.returnCode}")
            .doOnError {
                Timber.e("MQTT BROKER >>> DON'T CONNECTED WITH HIVEMQ CLIENT::: ${it.localizedMessage}")

fun subscribe(topicFilter: String): Flowable<String> {
        return this.client.subscribeStreamWith()
            .doOnSingle { suback ->
                Timber.i("MQTT BROKER >>> TOPIC $topicFilter SUBSCRIBE COMPLETED WITH CODES: ${suback.returnCodes}")
            .doOnNext { message ->
                Timber.i("MQTT BROKER >>> MESSAGE RECEIVED --- ${String(message.payloadAsBytes)}")
            .doOnError {
                Timber.e("MQTT BROKER >>> SUBSCRIBE TOPIC ERRROR ")
            .map {