mqttconnection and wificonnection: modules or examples?

[bartmichu]

Hello. I'm looking at OS210302 release notes and I'm happy to see mqttconnection and wificonnection modules. Was it intentional to put them in the examples directory instead of modules? It makes me a bit confused - are they modules or examples? :)

[phoddie]

Fair question. They started out as examples. But both are generally useful so perhaps they should migrate to modules.

The continuous Wi-Fi scanner is in a similar situation. It is very useful for doing more complete scans and for detecting when access point come and go. The $MODDABLE//examples/network/wifi/wifiscancontinuous/ directory has both the module implementation and the example app.

[bartmichu]

Thanks for clarification, it's exactly what I was thinking - too useful and generic to be buried in examples section.

[phoddie]

I don't disagree. I'm not quite certain where they fit under modules. I will think about it. Do you have an idea?

[louisvangeldrop]

If I may do a suggestion: in the IO module as a provider.

[louisvangeldrop]

I have a request. MQTT is heavily used for exchanging all kind of info between devices. MQTT publish is per transaction. The connection can be built once, publish the data and close the connection. The subscribe is more difficult. It has to be active as a server for 100% of the time. Hence error recovery is top priority and difficult to implement, since it affects Wifi and MQTT. I have noticed that a Wifi-connection can not always esablished for an ESP32 device. The only way to get the Wifi connection established is by switching off the device.
Is there a template that is able to reconnect to ESP32 Wifi in any case- even if a restart is needed?

[phoddie]

I'm not certain I understand.

The only way to get the Wifi connection established is by switching off the device.

It sounds like you are describing a bug in the ESP-IDF where it cannot reconnect to Wi-Fi. We did see this in previous versions of the ESP-IDF last year but it was resolve in ESP-IDF v3.3.2, I believe. What is the circumstance where you see Wi-Fi re-connection failures?

The subscribe is more difficult

How is this related to the Wi-Fi re-connection?

[louisvangeldrop]

It has to do with "start", "disconnect" loop.

let wifi=new WiFi ( {ssid,passwd},(msg)=>{ switch (msg){ case "start" : do nothing case Wifi.disconnected: restart } }

Sometimes I get this loop. There is never a connect or gotip event.

[phoddie]

@louisvangeldrop - thank you for the details.

You write that you get a "loop" and show "start" and "disconnected". That combination should never be a loop. "start" should appear once at launch and then it should be some combination of "disconnected", "connected", and "gotip." Would you share a trace of the loop you see?

FWIW – I don't think the Wi-Fi layer should force a restart because of Wi-Fi connection failures. That does not seem like a decisions it should make. It is up to the application. Instead, the Wi-Fi layer could detect that several reconnection attempts have failed and report that as an error to the application. The application could choose to restart.

As an experiment, I added two additional timeouts to the implementation. They force a disconnect and reconnection attempt when it takes to long to receive either the connect or gotip messages. That will help in some circumstances, but I cannot say if it will help what you are seeing. The implementation is in this Gist for you to try.

[louisvangeldrop]

The combination was: first a "start"-event and then a "disconnected" event. Never either "connect" or "gotip" events occurred. You are right that the application shoud decide to force a restart/deepsleep.

In the meantime maybe I have found the reason why this happened. I am using a M5stack/Atom as a RF433 transmitter allowing me to switch on/off all kind of devices. THe nice thing of this small device is that you connect a RF433 transmitter directly into the pins GND, VCC and pin 25. Very easy to assemble. But I didn't write a 0-value to pin 25 at the start. According to this item from Esspressif https://docs.espressif.com/projects/esp-idf/en/latest/esp32/api-reference/peripherals/adc.html#analog-to-digital-converter pin 25 is also used by Wifi. After I have changed the code to initialize the pin to zero, all my Wifi-problems were gone.

[phoddie]

Excellent. Pin conflicts like that are no fun to track down.

[louisvangeldrop]

Yes, you are 100% right. I also use the TTGO-LilyGo with Wifi and MQTT-subscribe. In that case I intentionaly avoided the ADC2 pins. I assumed that the M5Stack people were offering ADC2-free pins for usage. Next time I will read the documentation more carefully.

[louisvangeldrop]

I have a feature request for the mqtt-connection example. The nodejs-mqtt client has the default option to restore the subscriptions after a connection-failure with the mqtt-server. A very nice to have option I believe.

[phoddie]

It is convenient. However, it requires the implementation to maintain a list of subscriptions which consumes memory for the lifetime of the connection instance.

The onReady callback is invoked on each connection (the initial connection and all subsequent reconnections). If you subscribe to the necessary topics in your onReady code, you'll be set - and no additional memory required. Or is there another scenario you have in mind?

[louisvangeldrop]

In the nodejs mqtt-client, it is an option, that can be set. It opens the opportunity to skip the option in case of memory shortage.

[phoddie]

I understand that. It isn't difficult to implement. But, I don't understand the need here. Apologies if I'm overlooking the obvious. What problem does it solve that issuing the subscriptions from the onReady callback does not address?

[louisvangeldrop]

The major need for me is that with multiple onReady-callback calls, I can not use the async-await programming model.

[phoddie]

To implement an async-await programming style, I assume you've either modified the mqttconnection.js or wrapped it (perhaps subclassed it) in some way to allow you to await for message received via onMessage. I suppose you also have an initial await that corresponds to onReady so you know when the connection is established? It would help if you could share how that is done. Thanks.

[louisvangeldrop]

I do not await the onMessage callback. I use the Connectection class and in the onReady-callback I do a resolve().

export function setupMQTTSubscribeAsync(options: { server: string, subscribe: string, handler: (topic: string, body: ArrayBuffer) => void, wait?: number }) {
    return new Promise((res, rej) => {
        let mqttReady = false
        const wait = options?.wait ?? 5000
        let mqtt

        if (mqttReady === false) {
            mqttReady = true
            mqtt = new Client({
                host: options.server,
                id: `subscribe_"${Net.get("MAC")}`
                // ,timeout: wait
            });
            mqtt.on('ready', function () {
                mqttReady = true
                mqtt.subscribe(`${options.subscribe}`, options.handler);
                trace(`mqtt ready: ${options.server}\n`)
                res(mqttReady)
                return mqttReady
            })
            // mqtt.onMessage = (topic, message) => options.handler(topic, message)
            mqtt.on('close', function () {
                mqttReady = false
                trace('lost connection to server\n');
                rej(mqttReady)
                return mqttReady
            })

        }

    })
}

[phoddie]

Thanks for the snippet. I'm still not following entirely. I assume Client is Connection class exported by mqttconnection.js. That class doesn't implement an on method, and that seems relevant here. How does that get added?

Maybe a straightforward subclass of Connection can do what you need? Something like this:

class ResubscribingConnection extends Connection {
    #subscriptions = [];
    #onReady;
    subscribe(topic) {
        super.subscribe(topic);
        if (!this.#subscriptions.includes(topic))
            this.#subscriptions.push(topic);
    }
    unsubscribe(topic) {
        super.unsubscribe(topic);
        const index =  this.#subscriptions.indexOf(topic);
        if (index >= 0)
            this.#subscriptions.splice(index, 1);
    }
    get onReady() {
        return this.#resubscribeOnReady;
    }
    set onReady(value) {
        this.#onReady = value;
    }
    #resubscribeOnReady() {
        this.#onReady?.();
        this.#subscriptions.forEach(topic => super.subscribe(topic));
    }
}

[louisvangeldrop]

My apologies for sending the wrong code after a couple of changes. However the idea of the Connection-class is the same as original Connection-class.

The reason that I started this request is that I moved code from nodejs to the Moddable platform and expected the automatic resubscribe. This feature makes it easier to migrate from nodejs ( at raspberry pi) to moddbale (esp32)

[phoddie]

Without seeing how you are using the mqttconnection class, it isn't clear why resubscribing from 'onReady' is unworkable. I hesitate to add features without understanding why they are needed. Would you share enough of your code to make that clear? Thank you.

[louisvangeldrop]

Connection = mqttconnection

export class Client extends Connection{
    #subscriptions = new Set()
    #resubscribe

    constructor(options) {
        var { resubscribe, ...options } = options
        super(options)
        this.#resubscribe = resubscribe
        super.onReady = this.#ready
        super.onClose = this.#close
        super.onMessage = this.#message
    }
    #close = () => {
        trace('connection lost')
        // this.onClose?.()
    }

    #ready = () => {
        for (let topic of this.#subscriptions) {
            this.subscribe(topic)
        }
        this.onReady?.()
    }

    #message = (topic, data) => {
        this.onMessage?.(topic, data)
    }

    subscribe = (topic) => {
        this.#subscriptions.add(topic)
        super.subscribe(topic)
        // super.onMessage = callback
    }

    unsubscribe = (topic) => {
        this.#subscriptions.delete(topic)
        super.unsubscribe(topic)
    }

}

export function setupMQTTSubscribeAsync(options: { server: string, subscribe: string, handler: (topic: string, body: ArrayBuffer) => void, wait?: number }) {
    return new Promise((res, rej) => {
        let mqttReady = false
        const wait = options?.wait ?? 5000
        let mqtt

        if (mqttReady === false) {
            mqttReady = true
            mqtt = new Client({
                host: options.server,
                id: `subscribe_"${Net.get("MAC")}`
                // ,timeout: wait
            });
            mqtt.onReady = function () {
                mqttReady = true
                mqtt.subscribe(`${options.subscribe}`)
                mqtt.onMessage = options.handler
                trace(`mqtt ready: ${options.server}\n`)
                res(mqttReady)
                return mqttReady
            }
            // mqtt.onMessage = (topic, message) => options.handler(topic, message)
            mqtt.onClose = function () {
                mqttReady = false
                trace('lost connection to server\n');
                rej(mqttReady)
                return mqttReady
            }

        }

    })
}

const setup = async () => {
    try {
        // wifi = await wifiAsync(ssid, password)
        let msg = await wifi({ ssid, password })
        trace(`Wi-Fi: ${msg}\n`);
        await setupMDNSAsync(config.mdns ?? "weatherstation").then((msg) => trace(`${msg}\n`))
        await setupSNTPAsync(config.sntp ?? '0.nl.pool.ntp.org').then((msg) => trace(`${msg}\n`))
        await setupMQTTSubscribeAsync({ server: mqttServer, subscribe: mqttSubscribeTopic, handler: mqttHandler })
    }
    catch (e) {
        trace(`error: ${e}\n`)
        restart()
    } finally {
        return
    }
}

[phoddie]

Thanks, that helps. (And excellent use of Set!).

But, I'm not sure the resubscribe code works. Does it? The code to set the callbacks (for example, super.onReady = this.#ready) in Client are immediately overridden by the assignments to onReady and onClose in setupMQTTSubscribeAsync.

[louisvangeldrop]

If the class Client extends MQTT, the code works.

[phoddie]

To be clear, you are now referring to mqtt.js and not mqttconnection.js? I ask because above you wrote Connection = mqttconnection so I read the code assuming that.

[louisvangeldrop]

I have changed the Client-class into the the class you have suggested in ResubscribingConnection. Also I changed the onClose handling into a nop(). Now it works fine. I can switch off and on the mosquitto and it recovers and resubscribes without any problem.

Maybe a nice addition for the mqtt examples folder

[phoddie]

Cool. Congratulations!

I spent a little time this morning experimenting with this too. The approach avoids the need for promises but fits within your async code pattern. Here's the modified example:

const mqtt = new AsyncFriendlyClient({
    host: "test.mosquitto.org",
    onMessage(topic, body) {
        trace(`received "${topic}": ${String.fromArrayBuffer(body)}\n`);
    }
});

mqtt.subscribe("moddable/mqtt/example/#");

Timer.repeat(() => {
    if (!mqtt.ready)
        return;

    mqtt.publish("moddable/mqtt/example/date", (new Date()).toString());
    mqtt.publish("moddable/mqtt/example/random", Math.random());
 }, 1000);

The caller doesn't deal with the onReady or onClosed callbacks. The onMessage callback moves to the constructor just to be consistent with our transition to Ecma-419 style. subscribe and unsubscribe are safe to call whether or not connected, and are cached. mqttconnection provides an accessor to ready so clients can check before calling publish. Here's the class implementation (probably similar to yours):

class AsyncFriendlyClient extends Client {
    #subscriptions = new Set;
    #onMessage

    constructor(options) {
        super(options);
        this.#onMessage = options.onMessage
    }
    onReady() {
        this.#subscriptions.forEach(topic => this.subscribe(topic));
    }
    onMessage(topic, body) {
        this.#onMessage?.(topic, body);
    }
    subscribe(topic) {
        this.#subscriptions.add(topic);
        if (super.ready)
            super.subscribe(topic);
    }
    unsubscribe(topic) {
        this.#subscriptions.delete(topic);
        if (super.ready)
            super.unsubscribe(topic);
    }
}

It depends on adding the ready accessor to mqttconnection.js:

    get ready() {
        return this.#ready;
    }

Maybe a nice addition for the mqtt examples folder

Agreed.

[louisvangeldrop]

Peter, thanks for your efforts and help with this challenge. I appreciate your help and your patience, since you must be very busy, I presume.

[phoddie]

No problem. It is tangentially related to network protocol work we are doing in Ecma-419, so it is a good exercise.

[phoddie]

@louisvangeldrop - the latest commit of mqttconnection.js incorporates resubscribe. The example has been updated (slightly different than above). It uses the onReady and onClose callbacks to turn schedule/unschedule the timer that performs the periodic writes. The ready accessor is also there, so the timer could always be enabled and instead check ready before each write.

https://github.com/Moddable-OpenSource/moddable/blob/ccce25693eafbef90bfce7243719564f71a25213/examples/network/mqtt/mqttconnection/main.js#L21-L34

Resubscribe is optional and defaults to on. That might require a change in existing code that uses it -- they just need to set subscribe: false in the constructor's options object to avoid sending redundant subscribe requests to the server.