peterhinch / micropython-mqtt

A 'resilient' asynchronous MQTT driver. Recovers from WiFi and broker outages.

MicroPython Asynchronous MQTT

MQTT is an easily used networking protocol designed for IOT (internet of things) applications. It is well suited for controlling hardware devices and for reading sensors across a local network or the internet.

It is a means of communicating between multiple clients. A single server, also known as a broker, manages the network. Clients may include ESP8266, ESP32 and Pyboard D modules and other networked computers. Typical server hardware is a Raspberry Pi or other small Linux machine which may be left running 24/7. An effective PC server is mosquitto. Public brokers also exist.

MQTT Packets are passed between clients using a publish/subscribe model. They consist of a topic and a message string. Clients subscribe to a topic and will receive all packets published by any client under that topic.

The protocol supports three "quality of service" (qos) levels. Level 0 offers no guarantees. Level 1 ensures that a packet is communicated to the recipient but duplication can occur. Level 2 avoids duplication; it is not supported by the official driver or by this module. Duplicates can readily be handled at the application level.

Main README

Warning: firmware >= V1.22.0

V1.22.0 included a changed IDF version 5.0.4: on ESPx the package should be replaced with the latest version, otherwise recovery from an outage may not occur.

1. Contents

1. Contents
   1.1 Rationale
   1.2 Overview
   1.3 Project Status
   1.4 ESP8266 limitations
   1.5 ESP32 Issues
   1.6 Pyboard D
   1.7 Arduino Nano RP2040 Connect
   1.8 RP2 Pico W
   1.9 Limitations Please read this.
   1.10 MQTTv5 Which version should you use?
2. Getting started
   2.1 Program files
   2.2 Installation
   2.3 Example Usage Using the event interface.
   2.4 Usage with callbacks
3. MQTTClient class
   3.1 Constructor Describes the MQTT configuration dictionary.
   3.2 Methods
        3.2.1 connect
        3.2.2 publish
        3.2.3 subscribe
        3.2.4 unsubscribe
        3.2.5 isconnected
        3.2.6 disconnect
        3.2.7 close
        3.2.8 broker_up
        3.2.9 wan_ok
        3.2.10 dprint
   3.3 Class Variables
   3.4 Module Attribute
   3.5 Event based interface
   3.6 MQTTv5 Support
        3.6.1 Configuration and Migration from MQTTv3.1.1
        3.6.2 MQTTv5 Properties
        3.6.3 Unsupported Features
4. Notes
   4.1 Connectivity
   4.2 Client publications with qos == 1
   4.3 Client subscriptions with qos == 1
   4.4 Application Design
        4.4.1 Publication Timeouts
        4.4.2 Behaviour on power up
5. Non standard applications Usage in specialist and micropower applications.
   5.1 deepsleep
   5.2 lightsleep and disconnect
   5.3 Ultra low power consumption For ESP8266 and ESP32.
6. References
7. Connect Error Codes
8. Hive MQ A secure, free, broker.
9. The ssl_params dictionary Plus user notes on SSL/TLS.

1.1 Rationale

The official "robust" MQTT client has the following limitations.

1. It is unable reliably to resume operation after a temporary WiFi outage.

2. It uses blocking sockets which can cause execution to pause for arbitrary periods when accessing a slow broker. It can also block forever in the case of qos == 1 publications while it waits for a publication acknowledge which never arrives; this can occur on a WiFi network if an outage occurs at this point in the sequence.

3. This blocking behaviour implies limited compatibility with asynchronous applications since pending coroutines will not be scheduled for the duration.

4. Its support for qos == 1 is partial. It does not support retransmission in the event of a publication acknowledge being lost. This can occur on a WiFi network especially near the limit of range or in the presence of interference.

5. Its partial qos == 1 support and inability reliably to resume after a WiFi outage places a limit on the usable WiFi range. To achieve reliable operation a client must be well within range of the access point (AP).

6. As a synchronous solution it has no mechanism to support the "keepalive" mechanism of MQTT. This prevents the "last will" system from working. It also makes subscribe-only clients problematic: the broker has no means of "knowing" whether the client is still connected.

This module aims to address these issues, at the cost of significant code size. It has been tested on the following platforms.

1. ESP8266
2. ESP32, ESP32-S2 and ESP32-S3
3. Pyboard D
4. Arduino Nano Connect
5. Raspberry Pi Pico W

The principal features of this driver are:

1. Non-blocking operation for applications using uasyncio.
2. Automatic recovery from WiFi and broker outages.
3. True qos == 1 operation with retransmission.
4. Improved WiFi range because of its tolerance of poor connectivity.

It has the drawback of increased code size which is an issue on the ESP8266. Run as frozen bytecode it uses about 50% of the RAM on the ESP8266. On ESP32 and Pyboard D it may be run as a standard Python module.

1.2 Overview

This module provides a "resilient" non-blocking MQTT driver. In this context "resilient" means capable of reliable operation in the presence of poor WiFi connectivity and dropouts. Clearly during a dropout or broker outage communication is impossible but when connectivity resumes the driver recovers transparently.

Near the limit of WiFi range communication delays may be incurred owing to retransmissions and reconnections but nonblocking behaviour and qos == 1 integrity are maintained.

It supports qos levels 0 and 1. In the case of packets with qos == 1 retransmissions will occur until the packet has successfully been transferred. If the WiFi fails (e.g. the device moves out out of range of the AP) the coroutine performing the publication will pause until connectivity resumes.

The driver requires the asyncio library and is intended for applications that use it. It uses nonblocking sockets and does not block the scheduler. The design is based on the official umqtt library but it has been substantially modified for resilience and for asynchronous operation.

It is primarily intended for applications which open a link to the MQTT broker aiming to maintaining that link indefinitely. Applications which close and re-open the link (e.g. for power saving purposes) are subject to limitations detailed in Non standard applications.

Hardware support: Pyboard D, ESP8266, ESP32, ESP32-S3, ESP32-S2, Pico W and Arduino Nano RP2040 Connect.
Firmware support: Official MicroPython firmware V1.19 or later.
Broker support: Mosquitto is preferred for its excellent MQTT compliance.
Protocol: The module supports a subset of MQTT revision 3.1.1.

1.3 Project Status

Initial development was by Peter Hinch. Thanks are due to Kevin Köck for providing and testing a number of bugfixes and enhancements. Also to other contributors, some mentioned below.

Note that in firmware prior to 1.21 asyncio was named uasyncio.

18 Aug 2024 V0.8.1 Reconfigured as a Python package. Bugfix in V5 support.
9 Aug 2024 V0.8.0 Partial MQTTv5 support contributed by Bob Veringa.
15 Feb 2024 V0.7.2 Make compliant with firmware V1.22.0 and later.
12 Nov 2022 V0.7.0 Provide alternative callback-free Event interface.
2 Nov 2022 Rename config.py to mqtt_local.py, doc improvements.
8 Aug 2022 V0.6.6 Support unsubscribe (courtesy of Kevin Köck's fork).
11 July 2022 V0.6.5 Support RP2 Pico W
5 July 2022 V0.6.4 Implement enhancements from Bob Veringa. Fix bug where tasks could fail to be stopped on a brief outage. Subscription callbacks now receive bytearrays rather than bytes objects.
10 June 2022 Lowpower demo removed as it required an obsolete version of asyncio. Improved handling of clean_init (issue #40).
21 May 2022 SSL/TLS ESP8266 support contributed by @SooOverpowered: see tls8266.py.
22 Apr 2022 Support added for Arduino Nano RP2040 Connect. See note below.
2 Aug 2021 SSL/TLS on ESP32 has now been confirmed working. Reference.

1.4 ESP8266 limitations

The module is too large to compile on the ESP8266 and should be precompiled or preferably frozen as bytecode. On the reference board with mqtt_as frozen, the demo script range_ex reports 27.4K of free RAM while running. The code disables automatic sleep: this reduces reconnects at cost of increased power consumption.

Notes on the Sonoff Basic R3 may be found here.

1.5 ESP32 issues

Firmware must now be official firmware as described above. The Loboris port has been abandoned by its author and is no longer supported.

1.6 Pyboard D

The library has been tested successfully with the Pyboard D SF2W and SF6W. In testing it has clocked up eight weeks of continuous runtime and nearly 1M messages without failure or data loss.

1.7 Arduino Nano RP2040 Connect

NINA firmware must up to date otherwise MicroPython produces error messages. See this doc. Reading RSSI seems to break the WiFi link so should be avoided - the range_ex.py demo disables this on this platform.

1.8 RP2 Pico W

The mqtt_as code should be V0.6.5 or later to avoid very slow recovery from outages.

1.9 Limitations

The MQTT 3.1 protocol supports extremely long messages. On a microcontroller message length is limited by available RAM. The actual limit will depend on the platform and user code but it is wise to design on the basis of a maximum of around 1KiB.

Avoid unrealistic expectations of performance: latency can be significant, especially when using a TLS connection to a broker located on the internet. With a non-encrypted connection to a local broker it is feasible to use one MicroPython client to control another. I haven't measured latency but I would guess at ~100ms.

Some platforms - notably ESP32 - are unhelpful when dealing with gross errors such as incorrect WiFi credentials. Initial connection will only fail after a one minute timeout. Other platforms enable an immediate bail-out.

1.10 MQTTv5

The addition of MQTTv5 support does not affect existing applications which will run unchanged. It is expected that most microcontroller users will continue with MQTT V3.1.1. The use of MQTTv5 uses additinal RAM (~3KiB) and requires some knowledge of the protocol. See MQTTv5 Support for more details.

Contents

2. Getting started

2.1 Program files

The library is configured as a Python package and is installed to an mqtt_as directory. This is typically located on the Python path. Demo scripts are then run with (e.g.):

>>> import mqtt_as.range

The user-configured mqtt_local.py (see below) is located on the Python path.

Required file

1. __init__.py The main module.
2. mqtt_v5_properties.py Only required if using MQTTv5.

Required by demo scripts

1. mqtt_local.py Holds local configuration details such as WiFi credentials. Place on Python path (usually / or /lib/).

Test/demo scripts

The first two of these demonstrate the event interface. Others use callbacks.

1. range.py For WiFi range testing. Good general demo.
2. range_ex.py As above but also publishes RSSI and free RAM. See code comments for limitations on Pico W and Arduino nano connect.
3. clean.py Test/demo program using MQTT Clean Session.
4. unclean.py Test/demo program with MQTT Clean Session False.
5. main.py Example for auto-starting an application.
6. tls.py Demo of SSL/TLS connection to a public broker. This runs on a Pyboard D. Publishes every 20s and subscribes to same topic. Connection to this public broker, though encrypted, is insecure because anyone can subscribe.
7. tls8266.py SSL/TLS connection for ESP8266. Shows how to use keys and certificates. For obvious reasons it requires editing to run.

Test scripts for MQTTv5:

1. basic.py Demo of user properties under MQTTv5.

Bash scripts (may be run on PC to publish periodically):

1. pubtest Bash script illustrating publication with Mosquitto.
2. pubtest_v5 Bash script illustrates various publication properties.

Quick install

ESP8266: please read Installation. On other platforms the main module, demos 1 to 3 and the sample mqtt_local.py may be installed from a connected PC with :

$ mpremote mip install github:peterhinch/micropython-mqtt

For MQTTv5, demos may be added with

$ mpremote mip install github:peterhinch/micropython-mqtt/mqtt_as/v5

An alternative is to use mip at the REPL with WiFi connected:

>>> import mip
>>> mip.install("github:peterhinch/micropython-mqtt")

The Bash scripts pubtest and pubtest_v5 should be copied to the PC.

Configuration

The MQTT client is configured using a dictionary. An instance named config is defined in the MQTTClient class and populated with common default values. The user can populate this in any manner. The approach used in the test scripts is as follows. The main __init__.py module instantiates config with typical defaults. Then mqtt_local.py adds local settings common to all nodes, e.g. WiFi credentials and broker details. Finally the application adds application specific settings like subscriptions.

In a typical project mqtt_local.py will be edited then deployed to all nodes.

The ESP8266 stores WiFi credentials internally: if the ESP8266 has connected to the LAN prior to running there is no need explicitly to specify these. On other platforms, or to have the capability of running on an ESP8266 which has not previously connected, mqtt_local.py should be edited to provide them. This is a sample cross-platform file:

from mqtt_as import config

config['server'] = '192.168.0.10'  # Change to suit e.g. 'iot.eclipse.org'

# Required on Pyboard D and ESP32. On ESP8266 these may be omitted (see above).
config['ssid'] = 'my_WiFi_SSID'
config['wifi_pw'] = 'my_password'

Contents

2.2 Installation

The module is too large to compile on the ESP8266. It must either be cross compiled or (preferably) built as frozen bytecode: copy __init__.py to esp8266/modules in the source tree, build and deploy. Copy mqtt_local.py to the filesystem for ease of making changes.

On other platforms simply copy the Python source to the filesystem (items 1 and 2 above as a minimum).

If an application is to auto-run on power-up it can be necessary to add a short delay in main.py:

import time
time.sleep(5)  # Could probably be shorter
import range  # Your application

This is platform dependent and gives the hardware time to initialise.

2.3 Example Usage

The library offers two alternative ways to handle events such as the arrival of a message. One uses traditional callbacks. The following uses Event instances and an asynchronous iterator. If a PC client publishes a message with the topic foo_topic the topic and message are printed. The code periodically publishes an incrementing count under the topic result.

from mqtt_as import MQTTClient, config
import asyncio

# Local configuration
config['ssid'] = 'your_network_name'  # Optional on ESP8266
config['wifi_pw'] = 'your_password'
config['server'] = '192.168.0.10'  # Change to suit e.g. 'iot.eclipse.org'

async def messages(client):  # Respond to incoming messages
    # If MQTT V5is used this would read
    # async for topic, msg, retained, properties in client.queue:
    async for topic, msg, retained in client.queue:
        print(topic.decode(), msg.decode(), retained)

async def up(client):  # Respond to connectivity being (re)established
    while True:
        await client.up.wait()  # Wait on an Event
        client.up.clear()
        await client.subscribe('foo_topic', 1)  # renew subscriptions

async def main(client):
    await client.connect()
    for coroutine in (up, messages):
        asyncio.create_task(coroutine(client))
    n = 0
    while True:
        await asyncio.sleep(5)
        print('publish', n)
        # If WiFi is down the following will pause for the duration.
        await client.publish('result', '{}'.format(n), qos = 1)
        n += 1

config["queue_len"] = 1  # Use event interface with default queue size
MQTTClient.DEBUG = True  # Optional: print diagnostic messages
client = MQTTClient(config)
try:
    asyncio.run(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors

The code may be tested by running pubtest in one terminal and, in another, mosquitto_sub -h 192.168.0.10 -t result (change the IP address to match your broker).

2.4 Usage with callbacks

The alternative callback-based interface may be run as follows:

from mqtt_as import MQTTClient, config
import asyncio

# Local configuration
config['ssid'] = 'your_network_name'  # Optional on ESP8266
config['wifi_pw'] = 'your_password'
config['server'] = '192.168.0.10'  # Change to suit e.g. 'iot.eclipse.org'

def callback(topic, msg, retained, properties=None):  # MQTT V5 passes properties
    print((topic.decode(), msg.decode(), retained))

async def conn_han(client):
    await client.subscribe('foo_topic', 1)

async def main(client):
    await client.connect()
    n = 0
    while True:
        await asyncio.sleep(5)
        print('publish', n)
        # If WiFi is down the following will pause for the duration.
        await client.publish('result', '{}'.format(n), qos = 1)
        n += 1

config['subs_cb'] = callback
config['connect_coro'] = conn_han

MQTTClient.DEBUG = True  # Optional: print diagnostic messages
client = MQTTClient(config)
try:
    asyncio.run(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors

As above, testing is done by running pubtest in one terminal and, in another, mosquitto_sub -h 192.168.0.10 -t result (change the IP address to match your broker).

Contents

3. MQTTClient class

The module provides a single class: MQTTClient.

3.1 Constructor

This takes a dictionary as argument. The default is mqtt_as.config which is populated with default values listed below. A typical application imports this and modifies selected entries as required. Entries are as follows (default values shown in []):

WiFi Credentials

These are required for platforms other than ESP8266 where they are optional. If the ESP8266 has previously connected to the required LAN the chip can reconnect automatically. If credentials are provided, an ESP8266 which has no stored values or which has stored values which don't match any available network will attempt to connect to the specified LAN.

'ssid' [None]
'wifi_pw' [None]

MQTT parameters

'client_id' [auto-generated unique ID] Must be a bytes instance.
'server' [None] Broker IP address (mandatory).
'port' [0] 0 signifies default port (1883 or 8883 for SSL).
'user' [''] MQTT credentials (if required).
'password' [''] If a password is provided a user must also exist.
'keepalive' [60] Period (secs) before broker regards client as having died.
'ping_interval' [0] Period (secs) between broker pings. 0 == use default.
'ssl' [False] If True use SSL.
'ssl_params' [{}] See below.
'response_time' [10] Time in which server is expected to respond (s). See note below.
'clean_init' [True] Clean Session state on initial connection. (Ignored if MQTT V5 is in use).
'clean' [True] Clean session state on reconnection. (Known as Clean Start in MQTT V5).
'max_repubs' [4] Maximum no. of republications before reconnection is attempted.
'will' : [None] A list or tuple defining the last will (see below).

Interface definition

'queue_len' [0] If a value > 0 is passed the Event-based interface is engaged. This replaces the callbacks defined below with a message queue and Event instances. See section 3.5.

Callback based interface

This interface is optional. It is retained for compatibility with existing code. In new designs please consider the event based interface which replaces callbacks with a more asyncio-friendly approach.

'subs_cb' [a null lambda function] Subscription callback. Runs when a message is received whose topic matches a subscription. The callback must take three or four args, topic, message, retained and properties=None. The first two are bytes instances, retained is a bool, True if the message is a retained message. properties is a dict (or None) if MQTT V5 is in use.
'wifi_coro' [a null coro] A coroutine. Defines a task to run when the network state changes. The coro receives a single bool arg being the network state.
'connect_coro' [a null coro] A coroutine. Defines a task to run when a connection to the broker has been established. This is typically used to register and renew subscriptions. The coro receives a single argument, the client instance.

MQTT V5 extensions

See MQTTv5 Support
'mqttv5' [False]
'mqttv5_con_props' [None]

Notes

The response_time entry works as follows. If a read or write operation times out, the connection is presumed dead and the reconnection process begins. If a qos==1 publication is not acknowledged in this period, republication will occur. May need extending for slow internet connections.

The will entry defines a publication which the broker will issue if it determines that the connection has timed out. This is a tuple or list comprising [topic (string), msg (string), retain (bool), qos (0 or 1)]. If the arg is provided all elements are mandatory.

Clean sessions: If clean is set, messages from the server during an outage will be lost regardless of their qos level.

If clean is False messages sent from the server with qos == 1 will be received when connectivity resumes. This is standard MQTT behaviour (MQTT spec section 3.1.2.4). If the outage is prolonged this can imply a substantial backlog. On the ESP8266 this can cause buffer overflows in the Espressif WiFi stack causing LmacRxBlk:1 errors to appear. see this doc.

clean_init should normally be True. If False the system will attempt to restore a prior session on the first connection. This may result in a large backlog of qos==1 messages being received, for example if a client is taken out of service for a long time. This can have the consequences described above. See MQTT spec 3.1.2.4. This is described further below in section 4.4.2 behaviour on power up.

SSL/TLS

Populating the ssl_params dictionary is something of a black art. Some sites require certificates: see this post for details on how to specify these. See Hive MQ for details of connecting to a secure, free broker service. This may provide hints for connecting to other TLS brokers.

Contents

3.2 Methods

Note re data types. Messages and topics may be strings provided that all characters have ordinal values <= 127 (Unicode single byte characters). Otherwise the string encode method should be used to convert them to bytes objects.

3.2.1 connect

Asynchronous.

Keyword only arg:

• quick=False Setting quick=True saves power in some battery applications. It does this on (re)connect by skipping the check on WiFi integrity. This check is intended for mobile clients that may attempt to reconnect under conditions of poor signal strength. In conditions of good signal strength this check may be skipped.
  See Non standard applications.

Connects to the specified broker. The application should call connect once on startup. If this fails (due to WiFi or the broker being unavailable) an OSError will be raised: see Connect Error Codes. Subsequent reconnections after outages are handled automatically.

3.2.2 publish

Asynchronous.

If connectivity is OK the coro will complete immediately, else it will pause until the WiFi/broker are accessible. Section 4.2 describes qos == 1 operation.

Args:

1. topic A bytes or bytearray object. Or ASCII string as described above.
2. msg A bytes or bytearray object.
3. retain=False Boolean.
4. qos=0 Integer.
5. properties=None See MQTTv5 Support.

3.2.3 subscribe

Asynchronous.

Subscriptions should be created in the connect coroutine to ensure they are re-established after an outage.

The coro will pause until a SUBACK has been received from the broker, if necessary reconnecting to a failed network.

Args:

1. topic A bytes or bytearray object. Or ASCII string as described above.
2. qos=0 Integer.

It is possible to subscribe to multiple topics but there can only be one subscription callback.

3.2.4 unsubscribe

Asynchronous.

The coro will pause until an UNSUBACK has been received from the broker, if necessary reconnecting to a failed network.

Arg:

1. topic A bytes or bytearray object. Or ASCII string as described above.

If there is no subscription in place with the passed topic name the method will complete normally. This is in accordance with MQTT spec 3.10.4 Response.

3.2.5 isconnected

Synchronous. No args.

Returns True if connectivity is OK otherwise it returns False and schedules reconnection attempts.

3.2.6 disconnect

Asynchronous. No args.

Sends a DISCONNECT packet to the broker, closes socket. Disconnection suppresses the Will (MQTT spec. 3.1.2.5). This may be done prior to a power down or deepsleep. For restrictions on the use of this method see lightsleep and disconnect.

3.2.7 close

Synchronous. No args.

Shuts down the WiFi interface and closes the socket. Its main use is in development to prevent ESP8266 LmacRxBlk:1 failures if an application raises an exception or is terminated with ctrl-C (see Example Usage.

3.2.8 broker_up

Asynchronous. No args.

Unless data was received in the last second it issues an MQTT ping and waits for a response. If it times out (response_time exceeded) with no response it returns False otherwise it returns True.

3.2.9 wan_ok

Asynchronous.

Returns True if internet connectivity is available, else False. It first checks current WiFi and broker connectivity. If present, it sends a DNS query to '8.8.8.8' and checks for a valid response.

There is a single arg packet which is a bytes object being the DNS query. The default object queries the Google DNS server.

Please note that this is merely a convenience method. It is not used by the client code and its use is entirely optional.

3.2.10 dprint

If the class variable DEBUG is true, debug messages are output via dprint. This method can be redefined in a subclass, for example to log debug output to a file. The method takes an arbitrary number of positional args as per print.

3.3 Class Variables

1. DEBUG If True causes diagnostic messages to be printed.
2. REPUB_COUNT For debug purposes. Logs the total number of republications with the same PID which have occurred since startup.

3.4 Module Attribute

1. VERSION A 3-tuple of ints (major, minor, micro) e.g. (0, 5, 0).

3.5 Event based interface

This is invoked by setting config["queue_len"] = N where N > 0. In this mode there are no callbacks. Incoming messages are queued and may be accessed with an asynchronous iterator. The module reports connectivity changes by setting bound .up and .down Event instances. The demos range.py and range_ex.py use this interface. The following code fragments illustrate its use.

Reading messages:

async def messages(client):
    # If MQTT V5 is in use
    # async for topic, msg, retained, properties in client.queue:
    async for topic, msg, retained in client.queue:
        print(f'Topic: "{topic.decode()}" Message: "{msg.decode()}" Retained: {retained}')

Handling connect events:

async def up(client):  # (re)connection.
    while True:
        await client.up.wait()
        client.up.clear()
        print('We are connected to broker.')
        await client.subscribe('foo_topic', 1)  # Re-subscribe after outage

Optional outage handler:

async def down(client):
    while True:
        await client.down.wait()  # Pause until outage
        client.down.clear()
        print('WiFi or broker is down.')

Initialisation with a default small queue:

config["queue_len"] = 1
client = MQTTClient(config)

In applications where incoming messages arrive slowly and the clean flag is False, messages will not accumulate and the queue length can be small. A value of 1 provides a minimal queue. The queue comes into play when bursts of messages arrive too quickly for the application to process them. This can occur if multiple clients independently publish to the same topic: the broker may forward them to subscribers at a high rate. Another case is when the clean flag is False and a long wifi outage occurs: when the outage ends there may be a large backlog of messages. Such cases may warrant a larger queue.

In the event of the queue overflowing, the oldest messages will be discarded. This policy prioritises resilience over the qos==1 guarantee. The bound variable client.queue.discards keeps a running total of lost messages. In development this can help determine the optimum queue length.

It is possible (though seldom useful) to have multiple tasks waiting on messages. These must yield control after each message to allow the others to be scheduled. Messages will be distributed between waiting tasks in a round-robin fashion. Multiple instances of this task may be created:

async def messages(client):
    async for topic, msg, retained in client.queue:
        await asyncio.sleep(0)  # Allow other instances to be scheduled
        # handle message

In applications RAM is at a premium, in testing the callback-based interface offers somewhat (~1.3KiB) lower consumption than the minimal queue case.

Contents

3.6 MQTTv5 Support

Application designers should consider whether V5 is appropriate. Relevant factors include:

• V5 is essential if the client is to subscribe to messages published by a V5 client where the massage includes properties which must be received.
• V5 offers major benefits if the client is to run with clean session False. The intention here is to ensure that, where connectivity is unreliable, messages are not lost during brief outages. In V3.1.1 this has the unfortunate consequence that, where a prolonged outage occurs, there can be a large backlog. V5 messages may be published with a "message expiry interval" property, offering an option to limit the size of the backlog. This is further enhanced by a "session expiry interval" set on connection.
• V5 is essential if there is a requirement to publish messages with properties.
• Where RAM is limited, V5 adds about 3KiB to the library's usage.
• V5 enables a request/response protocol where an incoming message can be associated with a specific publication.

3.6.1 Configuration and Migration from MQTTv3.1.1

MQTTv5 is supported and can be configured by setting mqttv5 to True in the the config dictionary. The default is False. V5 applications should not use the clean_init config value - message backlogs should be controlled using session and message expiry intervals. Properties on connect are supported, these need to be passed in the configuration dictionary. See 3.6.2 for more information on properties, and how to format them.

from mqtt_as import MQTTClient, config
config['mqttv5'] = True

# Optional: Set the properties for the connection
config['mqttv5_con_props'] = {
    0x11: 3600,  # Session Expiry Interval
}

# The rest of the configuration
client = MQTTClient(config)

There are modifications to the API to support MQTTv5 features. The most significant is the addition of the properties argument that is provided as an additional argument to both the event and callback-based message handlers.

# For MQTT 3.1.1 support
def callback(topic, msg, retained):
    print((topic, msg, retained))

# For MQTT 5 and 3.1.1 support
def callback(topic, msg, retained, properties=None):
    print((topic, msg, retained, properties))

Allowing properties as an optional argument allows you to switch between MQTT 3.1.1 and MQTT 5 support without changing the callback signature.

async def messages(client):
    async for topic, msg, retained, properties in client.queue:
        await asyncio.sleep(0)  # Allow other instances to be scheduled
        # handle message

The properties argument is a dictionary that contains the properties of the message. The properties are defined in the MQTTv5 specification. If you include properties in published messages, while using MQTTv3.1.1, the properties will be ignored.

3.6.2 MQTTv5 Properties

Properties are a new and important feature of MQTTv5. They are used to provide additional information about the message, and allow for more advanced features such as message expiry, user properties, and response information.

Incoming properties are formatted as a dictionary using the property identifier as the key. The property identifier is an integer that is defined in the MQTTv5 specification, there are no constants defined in the module for these values. The property identifier is defined in the MQTTv5 specification.

Sending properties must be done in the right format. The MQTTv5 specification makes a distinction between binary and text properties. It is important to ensure that the properties are sent in the correct format. For reference, refer to section 2.2.2.2 of the MQTTv5 specification.

properties = {
    0x26: {'value': 'test'},       # User Property (UTF-8 string pair)
    0x09: b'correlation_data',     # Correlation Data (binary)
    0x08: 'response_topic',        # Response Topic (UTF-8 string)
    0x02: 60,                      # Message Expiry Interval (integer)
}

await client.publish('topic/test', 'message', False, 0, properties=properties)

In the following tables of properties types are defined as Python variable types; "string" is a utf8-encoded str. The V5 protocol provides for an optional request/response exchange. This is described in the V5 specification section 4.10

Outgoing properties

The following is a summary of properties relevant to client.publish().

Key	Value	Destination	Name	Meaning
0x01	byte	subscriber	payload format indicator	0=binary 1=utf8
0x02	int	broker	Message Expiry Interval	Lifetime in seconds
0x03	string	subscriber	Content Type	Application defined
0x08	string	subscriber	Response Topic	Request/response
0x09	bytes	subscriber	Correlation Data	Request/response
0x23	int	broker	topic alias
0x26	string pair	subscriber	user property	Application defined

Properties relevant to client.subscribe():

Key	Value	Name	Meaning
0x0B	int	Subscription Identifier	See below
0x26	string pair	user property	Application defined

The subscription identifier enables a client application to pass its current state to the broker: responses to that subscription will include that state. See spec section 3.8.4

Properties relevant to client.connect(). Note that connection properties are provided in the configuration dictionary (config[mqttv5_con_props]). All are instructions to the broker. All except 0x11 and 0x22 are somewhat esoteric and require study of the spec.

Key	Value	Name	Meaning
0x11	int	Session expiry interval (secs)	See below
0x17	byte 0/1	Request Problem Information	Spec 3.1.2.11.7
0x19	byte 0/1	Request Response Information	Spec 3.1.2.11.6
0x21	int	Receive Maximum	Spec section 4.9 etc
0x22	int	Topic Alias Maximum	Max alias value
0x26	string pair	user property	Application defined
0x27	int	Maximum Packet Size	Spec 3.2.2.3.6

The Session Expiry Interval defines how long the broker will retain session state after a disconnection. Setting it to zero and setting clean=True is equivalent to Clean Session status in MQTT3.1.1: publications occurring during the outage will be missed. A long expiry interval enables such messages to be received, at risk of a large backlog after a prolonged outage.

Incoming properties

Incoming message:

Key	Value	Source	Name	Meaning
0x01	byte	publisher	payload format indicator	0=binary 1=utf8
0x02	int	publisher	Message Expiry Interval	Lifetime in seconds
0x03	string	publisher	Content Type	Application defined
0x08	string	publisher	Response Topic	Request/response
0x09	bytes	publisher	Correlation Data	Request/response
0x0B	int	publisher	Subscription Identifier
0x26	string pair	publisher	user property	Application defined

Other packets received from the broker may contain properties. Apart from CONNACK the mosquitto broker seems to send these only under error conditions. In any event the client prints these if MQTTClient.DEBUG is True. Example packets are CONNACK, PUBACK, SUBACK, UNSUBACK and DISCONNECT.

Topic Alias

The idea behind the topic alias is to reduce outgoing message size. A publication is made with the full topic name with the topic alias set to a nonzero integer. Subsequent publications may pass a topic of "" with the topic alias property set to that integer. It is essential that the broker receives the message setting the alias as it can disconnect if an unknown alias is received. A problem also arises if an outage occurs while publishing aliased messages. It seems that the broker does not store aliases after an outage. It is the responsibility of the application to re-establish any aliases on reconnect. Please study the spec before using this feature.

3.6.3 Unsupported Features

In the interest of keeping the library lightweight and well tested, some features of MQTTv5 are not supported.

1. Enhanced Authentication: Enhanced Authentication is a new part of the MQTT specification that allows for more advanced authentication methods. This feature is not supported by the library. AUTH packet is not implemented and is not handled.
2. Will Properties: Will Properties with the introduction of properties in MQTTv5 messages can now have properties. This includes the will message. This feature is NOT supported, so properties cannot be sent with the will message.
3. Multiple User Properties: User Properties the spec allows for multiple user properties to be sent with a message. In the current implementation, only one user property is supported. This applied to both sending and receiving messages. When receiving messages, only the last user property is returned. If you include more than 1 key-value pair in the user properties dictionary when sending a message, only the first key-value pair will be sent.
4. Subscription options: Subscription Options in MQTTv5 subscription options were introduced (in addition to the QoS level). These options cannot be set when subscribing to a topic. The following options are not available:
   • No Local (NL)
   • Retain As Published (RAP)
   • Retain Handling
5. Not all properties in the CONNACK packet are exposed.
6. Properties on operations other than CONNECT and PUBLISH are not returned to the user. For more information, see this comment
7. The client does not store incoming Topic Alias properties.

NOTE: Most of these features could be implemented with some effort. These features were not implemented, to keep the current implementation simple and reduce the scope of testing required.

Contents

4. Notes

4.1 Connectivity

If keepalive is defined in the constructor call, the broker will assume that connectivity has been lost if no messages have been received in that period. The module attempts to keep the connection open by issuing an MQTT ping up to four times during the keepalive interval. (It pings if the last response from the broker was over 1/4 of the keepalive period). More frequent pings may be desirable to reduce latency of outage detection. This may be done using the ping_interval configuration option. The point here is that while WiFi failures are detected fast, upstream failure can only be detected by an absence of communication from the broker. With a long ping interval, the broker could be unreachable for a long time before the client detects it and initiates a reconnection attempt.

If the broker times out it will issue the "last will" publication (if any). This will be received by other clients subscribed to the topic.

If the client determines that connectivity has been lost it will close the socket and periodically attempt to reconnect until it succeeds.

In the event of failing connectivity client and server publications with qos == 0 may be lost. The behaviour of qos == 1 packets is described below.

4.2 Client publications with qos 1

These behave as follows. The client waits for response_time. If no acknowledgment has been received it re-publishes it, up to MAX_REPUBS times. In the absence of acknowledgment the network is presumed to be down. The client reconnects as described above. The publication is then attempted again as a new message with a different PID. (The new PID proved necessary for Mosquitto to recognise the message).

This effectively guarantees the reception of a qos == 1 publication, with the proviso that the publishing coroutine will block until reception has been acknowledged.

It is permissible for qos == 1 publications to run concurrently with each paused pending acknowledgement, however this has implications for resource constrained devices. See Section 4.4.

4.3 Client subscriptions with qos 1

Where the client is subscribed to a topic with qos == 1 and a publication with qos == 1 occurs the broker will re-publish until an acknowledgment is received. If the broker deems that connectivity has failed it waits for the client to reconnect. If the client was configured with clean set True, qos == 1 messages published during the outage will be lost. Otherwise they will be received in quick succession (which can overflow the buffer on an ESP8266 resulting in LmacRxBlk:1 messages).

4.4 Application design

The module allows concurrent publications and registration of subscriptions.

When using qos == 1 publications on hardware with limited resources such as ESP8266 it is wise to avoid concurrency by implementing a single publication task. In such cases if a publication queue is required it should be implemented by the application.

On capable hardware it is valid to have multiple coroutines performing qos == 1 publications asynchronously, but there are implications where connectivity with the broker is slow: an accumulation of tasks waiting on PUBACK packets implies consumption of resources.

The WiFi and Connect coroutines should run to completion quickly relative to the time required to connect and disconnect from the network. Aim for 2 seconds maximum. Alternatively the Connect coro can run indefinitely so long as it terminates if the isconnected() method returns False.

The subscription callback will block publications and the reception of further subscribed messages and should therefore be designed for a fast return.

4.4.1 Publication Timeouts

A contributor (Kevin Köck) was concerned that, in the case of a connectivity outage, a publication might be delayed to the point where it was excessively outdated. He wanted to implement a timeout to cancel the publication if an outage caused high latency.

Simple cancellation of a publication task is not recommended because it can disrupt the MQTT protocol. There are several ways to address this:

1. Send a timestamp as part of the publication with subscribers taking appropriate action in the case of delayed messages.
2. Check connectivity before publishing. This is not absolutely certain as connectivity might fail between the check and publication commencing.
3. Subclass the MQTTClient and acquire the self.lock object before issuing the cancellation. The self.lock object protects a protocol sequence so that it cannot be disrupted by another task. This was the method successfully adopted and can be seen in mqtt_as_timeout.py.

This was not included in the library mainly because most use cases are covered by use of a timestamp. Other reasons are documented in the code comments.

4.4.2 Behaviour on power up

The library aims to handle connectivity outages transparently, however power cycling of the client must be considered at application level. When the application calls the client's connect method any failure will cause an OSError to be raised. This is by design because the action to be taken is application-dependent. A check on WiFi or broker function may be required. There may be a need to fall back to a different network. In other applications brief power outages may be expected: when power resumes the client will simply reconnect. If an error occurs the application might wait for a period before re-trying.

When that initial connection has been achieved, subsequent connections caused by network outages are handled transparently to the application.

The behaviour of "clean session" should be considered in this context. If the clean flag is False and a long power outage occurs there may be a large backlog of messages. This can cause problems on resource constrained clients, notably if the client has been taken out of service for a few days. This module addresses this by enabling behaviour which differs between the power up case and the case of a network outage.

The clean_init flag determines behaviour on power up, while clean defines behaviour after a connectivity outage. If clean_init is True and clean is False, on power up prior session state is discarded. The client reconnects with clean==False. It reconnects similarly after connectivity outages. Hence, after power up, subscriptions will meet the qos==1 guarantee for messages published during connectivity outages.

If both flags are False normal non-clean behaviour ensues with the potential for substantial backlogs after long power outages.

If on power up both flags are True the broker will discard session state during connectivity (and hence power) outages. This implies a loss of messages published during connectivity outages(MQTT spec 3.1.2.4 Clean Session).

Also discussed here.

Contents

5. Non standard applications

Normal operation of mqtt_as is based on attempting to keep the link up as much as possible. This assures minimum latency for subscriptions but implies power draw. The machine module supports two power saving modes: lightsleep and deepsleep. Currently asyncio supports neither of these modes. The notes below may be relevant to any application which deliberately closes and re-opens the link to the broker.

5.1 deepsleep

Maximum power savings may be achieved by periodically connecting, handling publications and pending subscriptions, and entering deepsleep. With suitable hardware it is possible to produce an MQTT client with very low average power consumption. This is done by keeping the application run time short and using machine.deepsleep to sleep for a period. When the period expires the board resets and main.py re-starts the application.

Hardware tested was the UM Feather S2 available from Adafruit. My sample consumes only 66μA in deepsleep mode. It has a switchable LDO regulator allowing external sensors to be powered down when the host is in deepsleep. It also supports battery operation via a LiPo cell with USB charging. A Pyboard D with WBUS-DIP28 has similar properties.

The test script lptest_min.py wakes up periodically and connects to WiFi. It publishes the value from the onboard light sensor, and subscribes to the topic "foo_topic". Any matching publications which occurred during deepsleep are received and revealed by flashing the blue LED.

Note that deepsleep disables USB. This is inconvenient in development. The script has a test mode in which deepsleep is replaced by time.sleep and machine.soft_reset keeping the USB link active. An alternative approach to debugging is to use a UART with an FTDI adaptor. Such a link can survive a deep sleep.

Each time the client goes into deepsleep it issues .disconnect(). This sends an MQTT DISCONNECT packet to the broker suppressing the last will as per MQTT spec para 3.1.2.5. The reasoning is that deepsleep periods are likely to be much longer than the keepalive time. Using .disconnect() ensures that a last will message is only triggered in the event of a failure such as a program crash.

In applications which close the connection and deepsleep, power consumption may be further reduced by setting the quick arg to .connect. On connecting or re-connecting after an outage a check is made to ensure that WiFi connectivity is stable. Quick connection skips this check on initial connection only, saving several seconds. The reasoning here is that any error in initial connection must be handled by the application. The test script sleeps for retry seconds before re-trying the connection.

5.2 lightsleep and disconnect

The library is not designed for use in cases where the system goes into lightsleep. Firstly asyncio does not support lightsleep on all platforms - notably on STM where the ticks_ms clock (crucial to task scheduling) stops for the duration of lightsleep.

Secondly the library has no mechanism to ensure all tasks are shut down cleanly after issuing .disconnect. This calls into question any application that issues .disconnect and then attempts to reconnect. This issue does not arise with deepsleep because the host effectively powers down. When the sleep ends, asyncio and necessary tasks start as in a power up event.

These problems have been resolved by users for specific applications with forks of the library. Given the limitations of asyncio I do not plan to write a general solution.

5.3 Ultra low power consumption

This document describes an MQTT client for ESP32 or ESP8266 which uses ESPNOw to communicate with a gateway running mqtt_as. The client does not need to connect to WiFi each time it wakes, saving power. The gateway can be shared between multiple clients.

Drawbacks are the need for an always-on gateway, and the fact that only a subset of MQTT capabilities is supported.

Contents

6. References

mqtt introduction
mosquitto server
mosquitto client publish
mosquitto client subscribe
MQTT 3.1.1 spec
MQTTv5 spec
python client for PC's
Unofficial MQTT FAQ
List of public brokers

Contents

7. Connect Error Codes

On the initial connection attempt the broker may reject the attempt. In this instance an OSError will be raised showing two numbers. The first number should be 0x2002 which is the MQTT CONNACK fixed header. The second is CONNACK variable header byte 2 which indicates the reason for failure as follows:

Value	Reason
1	Unacceptable protocol version.
2	Client identifier rejected.
3	MQTT service unavailable.
4	Username or password have an invalid format.
5	Client is not authorised to connect.

See MQTT spec section 3.2.2.

Contents

8. Hive MQ

The Hive MQ site offers a free web-based broker which is more secure than public brokers. With a public broker anyone can detect and subscribe to your publications. Hive MQ gives you a unique broker internet address which requires a password to access. TLS is mandatory but does not require certificates.

A simple GitHub registration gets you:

• The unique broker address.
• You specify a username.
• The website supplies a password.

Typical usage:

config['user'] = 'my_username'
config['password'] = 'my_password'
broker = 'unique broker address'  # e.g long_hex_string.s2.eu.hivemq.cloud
config['server'] = broker
config['ssl'] = True
config['ssl_params'] = {"server_hostname": broker}

The free service is scalable (at cost) to large commercial deployments.

Contents

9. The ssl_params dictionary

The following are the allowable keys:

• 'key'
• 'cert'
• 'server_side'
• 'server_hostname'
• 'do_handshake'
• 'cert_reqs' mbedtls only
• 'cadata' mbedtls only

According to this post the following platforms use mbedtls:

• esp32 port
• pico w
• unix port
• stm32

See this post for details of how to use client certificates with a mosquitto broker.

Note also that TLS with client certificates requires the cliet's clock to be approximately correct. This can be achieved with an NTP query. If mosquitto is run on a local server it also runs the NTP daemon. A high availability option is to run the NTP query against the local server. See this doc, also the official ntptime module.

See this link for information on creating client certificates and a Bash script for doing so.

micropython-mqtt's Issues

Broken compatibility to ESP32_Lobo

In line 469 the network status is compared to "network.STAT_CONNECTING" but Loboris fork doesn't have this implemented, just like the pyboard.

why republish with new PID?

In the mqtt_as readme you write:

The publication is then attempted again as a new message with a different PID. (The new PID proved necessary for Mosquitto to recognise the message).

Can you describe why a new PID was necessary? If the client reconnects it would be best not to use a new PID to avoid unnecessary duplication of the message if the broker had actually received it. Is there a bug in mosquitto or what did you observe?

Edit: maybe I know what happens: when MQTTClient.publish calls MQTT_base.publish to republish the dup field is set to 0 and mosquitto probably barfs at that if it has indeed seen the packet before. IIUC dup needs to be set to 0 even if republishing on a fresh connection.

For mqtt broker is error, the noblock sock alway catch 'EINPROGRESS'

maybe set the socket as block to catch the error of connection, then set as noblock.

` async def _connect(self, clean):
# during connecting, set as noblocking to advoid err 'EINPROGRESS'
while True:
try:
self._sock = socket.socket()
self._sock.settimeout(3.0)
self._sock.connect(self._addr)
break
except OSError as e:
print("sock.connect:",e)
self._sock.close()
await asyncio.sleep_ms(8000)

   self._sock.settimeout(0)  `

Use Mqtt_as from inside a class

Hello Peter,
I unsuccessfully tried to implement the MQTT class into an own class to control it.
But so far it is not working and I don't understand why.
It works to publish messages from inside a class but not to subscribe to topics or calling the callback-func.

This is the Full code. which should make it self-explanatory what I'm trying to archive (I hope).

Comment Notes:

1. MQTTClient.LOGGING = True is the logging functionality described in the pull request.
2. The Async callback works after the first pull request (more details there). BUT I also tried it with the master version of mqtt_as to make sure it is not conflicting.
3. using the client in the class via self.client would make it obsolete to have a client argument but if i remove it, it will raise a TypeError, because it has this (self) argument in Line 511.
   

   micropython-mqtt/mqtt_as/mqtt_as.py

   Line 511 in 73cac20

   loop.create_task(self._connect_handler(self)) # User handler.

Also: client.connect(), self.client.publish works.
Any Help would be highly appreciated!

from mqtt_asx import MQTTClient
from config import config
import uasyncio as asyncio
import asyn
import logging

logging.basicConfig(level=logging.DEBUG)
log_mqtt = logging.getLogger("Mqtt")

SERVER = 'x.x.x.x'
DEVICE = "blind1"

# Sub topics
sub_cmd = "OPENBlind/{}/cmd/#".format(DEVICE)
sub_set = "OPENBlind/{}/settings/#".format(DEVICE)
# Publish topics
pub_sta = "OPENBlind/{}/state".format(DEVICE)
pub_cmd_pos_res = "OPENBlind/{}/cmd/pos/result".format(DEVICE)
pub_set_res = "OPENBlind/{}/settings/result".format(DEVICE)

class Mqtt():

    def __init__(self, motor, state):
        # # # # # # # # # # # #
        self.motor = motor
        self.state = state
        self.config = config
        # # # # # # # # # # # #
        self.config['subs_cb'] = self.callback
        self.config['connect_coro'] = self.conn_hnd
        self.config['server'] = SERVER
        # # # # # # # # # # # #
        MQTTClient.LOGGING = True # Note 1: Added Logging functionality -> (see pullreq)
        # MQTTClient.DEBUG = True
        self.client = MQTTClient(self.config)
        loop = asyncio.get_event_loop()
        log_mqtt.debug("Mqtt class Initiated")
        # # # # # # # # # # # #
        try:
            loop.run_until_complete(self.main())
        finally:
            self.client.close()  # Prevent LmacRxBlk:1 errors

    async def callback(self, topic, msg): # Note 2: Async Callback works after (See pullreq)
        log_mqtt.debug("Callback: {}, {}".format(topic,msg))
        await asyncio.sleep(1)
        print((topic, msg))

    async def conn_hnd(self, client): # Note 3: TypeError: line 511 function takes 1 positional arguments but 2 were given (self argument)
        await self.client.subscribe(sub_cmd, 1)
        await self.client.subscribe(sub_set, 1)
        log_mqtt.debug("Subscripedt To: CMD, SET")

    async def main(self):
        await self.client.connect()  # Con. from class works!
        # await self.conn_hnd(self.client) # <-- Note: i tried to start it manualy wich worked but did not call the callback. (with or without async.)
        # n = 0
        log_mqtt.debug("conected To client")
        await self.client.publish(pub_sta, 'Online', qos=1) # Pub. from class works!
        while True:
            pass

    def check_state(self):
        _, a, b = self.state()
        if a or b == 1:
            return False
        else:
            return True

Contact you? trying Your code on ESP32

Hi sorry to contact you that way.
I'm trying out your umqtt asyncio code on esp32 and would like to contact you. If you are in any IRC Chat oder Skype or something else, could you let me know?
I'm not a python professional but with some trial and error i got a dht22 sensor and Oled Display to run and make the esp32 report to Home Assistant with mqtt simple.
After some tests i figured out, that if the broker goes down for some reason, the mqtt simple fails.
With your code this works. But when i simulated a wlan shutdown, my esp32 froze at reconnecting.
On Skype you could find me as w-e-s-s-i-x-1 (without the "-").

Problem with Wifi disconnecting every few minutes with Reason 8 ASSOC_LEAVE on ESP32

I have a problem where the wifi connection of the ESP32 (M5 Atom)
is lost every few minutes.
I read a lot posts in the expressif forum and the esp32 forum and tried
several things (DHCP lease, deactivating AP_IF, ...) but no success so far.

The reason stated in the console is

wifi: STA_DISCONNECTED, reason:8

which is

ASSOC_LEAVE
Disassociated, because the sending STA is leaving (or has left) BSS.

For debugging I added some things to the wifi_coro callback

async def wifi_han(state):

    log.debug('wifi_han called with state {0}'.format(state))
    log.debug('wlan isconnected {0}, mqtt isconnected {1}'.format(wlan.isconnected(), client.isconnected()))

    if state:
        log.info('WiFi is up.')
    else:
        log.info('WiFi is down (outages: {0})'.format(outages))

And the log right before the connection loss is always like

wifi_han called with state False
wlan isconnected True, mqtt isconnected False
WiFi is down
wifi:state: run -> init (0)
wifi:pm stop, total sleep time: 91068370 us / 97278642 us
wifi:new:<6,0>, old:<6,0>, ap:<255,255>, sta:<6,0>, prof:1
wifi: STA_DISCONNECTED, reason:8
MOD: check status network.STAT_CONNECTING, sleep

Afterwards there is a normal reconnect.

Now that looks strange to me, because the callback is called with False in _reconnect().

And calling the isconnected() functions for Wifi wlan.isconnected() and MQTT client client.isconnected() in that callback tells me

wlan isconnected True, mqtt isconnected False

This means that MQTT is trying to reconnect, thinks it has lost the network connection, but when I ask the Wifi function it seems to think that the network connection is still alive.

I have no clue on how to interpret this.
Could it be that there is no real reason for MQTT to think that the connection was lost but trigger a reconnect anyway?

Everything I found so far indicates that the ESP32 is disconnecting itself, not being kicked out by the router.
The only entries in the router log are that the ESP32 is disconnecting normally and connecting again. There are many other devices working fine running on that router, e.g. an ESP8266 and some Raspberries.

Any ideas?

Asynchronous callbacks

Hello, thank you for the effort put in this library. It allows me to subscribe and manage a wide set of topics, but there is something I've been struggling with as a callback for the MQTT_client can't be asynchronous, so that it cannot contain asynchronous methods. I find this quite strange, as the rest of the library is completely async, so I've made a slight change on it to allow async calls inside the callback, just writing an await in mqtt_as.py when calling the callback function, as seen here:

...
447.    msg = await self._as_read(sz)
448.    await self._cb(topic, msg)
449.    if op & 6 == 2:
...

I don't know if this was on purpose or if it was a mistake that would not bee noticeable in simpler callbacks as in the example provided in the README, but it might be nice to have it in the official code.

need pure async mqtt based on native asyncio StreamReader/Writer

hi Peter,

Thaks for very documented async samples. Very usefull !
Under ESP32 everethig is OK, but i work with ESP8266 and it has less RAM.

In my project i use native asyncio.start_server() with StreamReader / StreamWriter for stream handling.
Do you have pure async mqtt code without WiFi connect, Lamp flashing etc?
It would be good to reuse StreamReader/Writer from asyncio to decrease code in Ram

Tutorial installation needs clarification

I'm following the tutorial to do the installation on a ESP8266, and I'm stumped about what I need to do. I tried copying mqtt_as.py and config.py, and got a memory error which was expected given the warning.

Could you provide further reference or pointers about how to cross-compile the module or build it as frozen bytecode?

The module is too large to compile on the ESP8266. It must either be cross compiled or (preferably) built as frozen 
bytecode: copy mqtt_as.py to esp8266/modules in the source tree, build and deploy. Copy config.py to the filesystem 
for convenience.

When I change the ssid it keeps connecting to the old network if it is still in range

The function wifi_connect() has the line:

 s.connect()  # ESP8266 remembers connection.

Only after this times out does it then call:

s.connect(self._ssid, self._wifi_pw)

Is there a way to check the name of the connection that the ESP8266 remembered, and compare it to the ssid that the programmer has asked for before assuming that it hasn't changed?

mDNS capability

This library is practically the whole justification of the async system, because you can make a sensor with an oled display do its stuff as soon as you turn it on, while all the nonsense of wifi connections and mqtt-broker finding is carried out in the background.

But could it also be upgraded to implement mDNS as well, as a further stage in its login? I've done my first hack of at this (sending and receiving UDP packets and decoding them) here:

micropython/micropython-esp32#201 (comment)

This would be really useful because I have a lot of these ESP32s running this asynchronous mqtt code, and it's nice to be able to connect them to different networks (whose ssid and passwords I can set) where they can find the mqtt.local broker reliably without each individually needing to be reprogrammed with the new ipnumber.

MQTT Broker on microPython.

Hello.

Are there a MQTT Broker implementation for the microPython?

I want to use ESP32 as the WiFi Gateway and I would like to use a MQTT Broker on it.

Thank you.

Cannot configure uasyncio to support micropython-mqtt

Hi there,

Was excited to see this library and was moving towards using as a substitute for https://github.com/micropython/micropython-lib/tree/master/umqtt.simple with more robustness. However, I can't seem to get it to connect and respond at all. This seems to do with not having the right uasyncio setup but it's not obvious where this is documented.

I added the mqtt_as.py file in the ports/esp8266/modules folder and then built an image from source, taking care to put the uasyncio directory from micropython-lib in there as a frozen module too.

After flashing a Wemos D1 Mini with this new image like...
esptool.py --port /dev/ttyUSB0 --baud 1500000 write_flash --flash_size=32m 0 build/firmware-combined.bin
...the library successfully imports as expected. I then used ampy to put a suitable main.py in place.

However, I couldn't initially recreate similar topic-publish-acknowledgement behaviour as my successful test case...
https://github.com/cefn/retrotextual/blob/master/gist/mqtttest.py using the reference code https://github.com/peterhinch/micropython-mqtt/blob/master/mqtt_as/range.py

My modified version of range.py is mqttpulse.py and it simply changes the server and topic to match the test case I am already successfully running for umqtt.simple (which also relies on the implicit cached connection to Wifi, and is running in exactly the same deployed configuration).

The umqtt.simple mqtttest.py version responds by echoing the topic and message to serial, but the range.py version is non-responsive after boot, regardless of trying to trigger its message acknowledgement by calling...

mosquitto_pub --topic 'hello' --message 'world' --retain

...which successfully triggers a response from umqtt.simple

On boot my mqtttest.py initially reported the following...

ets_task(40100130, 3, 3fff837c, 4)
scandone
state: 0 -> 2 (b0)
WebREPL daemon started on ws://192.168.4.1:8266
WebREPL daemon started on ws://0.0.0.0:8266
Started webrepl in manual override mode
Traceback (most recent call last):
  File "main.py", line 16, in <module>
ImportError: no module named 'config'

MicroPython v1.9.3 on 2017-12-01; ESP module with ESP8266
Type "help()" for more information.
>>> state: 2 -> 0 (2)
reconnect
scandone
state: 0 -> 2 (b0)
state: 2 -> 3 (0)
state: 3 -> 5 (10)
add 0
aid 7
cnt 

connected with Kitchen2, channel 11
dhcp client start...
ip:192.168.43.201,mask:255.255.255.0,gw:192.168.43.1

I then realised that the config.py file needed to be placed in the modules directory too.

After putting that in place, I received a boot error (from uasyncio?), extracted from the boot sequence as below...

  File "main.py", line 24, in <module>
AttributeError: 'module' object has no attribute 'get_event_loop'

To fix this, I populated the path uasyncio/core.py in modules with the file from...

https://github.com/micropython/micropython-lib/blob/master/uasyncio.core/uasyncio/core.py

However, it doesn't change the error.

Switching to import uasyncio.core as asyncio fixes the error superficially, but just creates a new one as uasyncio.core doesn't have an implementation of the sleep() coroutine, so I get...

AttributeError: 'module' object has no attribute 'sleep'

...instead.

What's the way of unpicking this, as there doesn't seem to be any guidance at...

https://github.com/micropython/micropython-lib/tree/master/uasyncio

...and the top level guidance for micropython-mqtt simply says...

The only dependency is uasyncio from the MicroPython library. Ensure this is installed on the device.

Do not require SSID on ESP32

I can use a wifimanager or Ethernet interface, but MQTT client requires to configure wifi in it's config.

It should be optional even on ESP32.

exception in `await client.connect()`

Normally my application using mqtt_as runs fine. But today it raised the following exception:

E (34474) wifi:sta is connecting, return error
Task exception wasn't retrieved
future: <Task> coro= <generator object 'main' at 3ffe7140>
Traceback (most recent call last):
  File "uasyncio/core.py", line 1, in run_until_complete
  File "main.py", line 65, in main
  File "mqtt_as/mqtt_as.py", line 516, in connect
  File "mqtt_as/mqtt_as.py", line 489, in wifi_connect
OSError: Wifi Internal Error

micropython-mqtt/mqtt_as/mqtt_as.py

Line 489 in e97b1e6

s.connect(self._ssid, self._wifi_pw)

I run micropython v1.17 on a ESP32

Option in client.wifi_connect() to force/keep connections on initial call

If I start the library (by calling client.connect()) when the wifi hotspot is not turned on, then client.wifi_connect() line 489 raises an OSError.

However, if the wifi hotspot goes down after a connection has been made, then the client._keep_connected() task will trap this exception and retry forever without throwing an OSError.

It would be useful to have devices where the behavior doesn't change just because there's been an interruption in power -- ie the feature where they reconnect as and when the hotspot comes available doesn't depend on the accidental fact that the hotspot was there when the power came on.

Maybe this could be done with a client.connect(timeout=15) parameter that hangs on the connection forever if timeout=0.

(I am working on an ESP8266 with the mqtt_as library compiled in as frozen bytecode)

Socket not closed if first connect fails/wrong authentication

I took a careful look at the connect function which raises an error if the first connection to the broker fails, which could be due to not being able to connect to wifi, broker offline or wrong authentication.
This means that there is a possibility that the socket remains open. Since I have to build a loop around the first client.connect as the broker might just be down at the moment my device (re)starts, I would have to close the socket myself using:

while True:
    try:
        await client.connect()
    except OSError:
        client.close()
        await asyncio.sleep(5)

This is of course acceptable but a bit unexpected because if I can't connect, I don't expect to close the client because it is not connected.
You document in the README that client.connect() raises an OSError if the connection fails but do not mention that client.close() might be needed and the example code doesn't cover the possibility that client.connect() raises an OSError.

A different approach would be to change the client.connect() method to:

    async def connect(self):
        if not self._has_connected:
            await self.wifi_connect()  # On 1st call, caller handles error
            # Note this blocks if DNS lookup occurs. Do it once to prevent
            # blocking during later internet outage:
            self._addr = socket.getaddrinfo(self.server, self.port)[0][-1]
        self._in_connect = True  # Disable low level ._isconnected check
        clean = self._clean if self._has_connected else self._clean_init
        try:
            await self._connect(clean)
        except Exception:
            self.close()
            raise

This way the socket is always cleaned up and the user doesn't need to worry about it.

I understand that this might be a minor issue because it will only be a problem if the restart of a device happens during an outage of the broker or if the device has wrong credentials (and did not previously connect successfully).

wifi_connect() times out for ESP8266 but not for ESP32

After reporting Issue #29, I began to depend on wifi_connect() to timeout when there is no access point.

Unfortunately this doesn't work on the ESP32.

The problem is around line 512 of mqtt_as.py where it hangs on

while s.status() == network.STAT_CONNECTING:
        await asyncio.sleep(1)

The problem is that, while the ESP8266 gives up after 10 seconds and reports (s.status() == network.STAT_NO_AP_FOUND), the ESP32 never gives up and returns a different state. Therefore we are getting inconsistent behavior.

I've broken out the problematic code here so the behavior can be tested in isolation:

>>> import network, time
>>> w = network.WLAN()
[junk omitted]
>>> w.active(1)
>>> w.connect("BV6000", "password")
>>> time.sleep(300)   # give it five minutes!
... [lots of the following message]
I (108937) wifi: STA_DISCONNECTED, reason:201
no AP found
>>> w.status(), network.STAT_CONNECTING 
1000, 1001

Looking through the special cases in the code, I see that if the library thinks it has esp32_LoBo then it times out after 10 seconds, so I can get that feature by doing the following:

import mqtt_as
mqtt_as.LOBO = True

int topic

Although the docs do mention "string topic", the in-built MQTT.simple allows integer topics as well, but this leads to an error with mqtt_as.

purpose of max_repubs?

In mqtt_as' MQTT_base you have a _max_repubs config variable which defaults to 4. It governs how often publish republishes a message on the existing connection (https://github.com/peterhinch/micropython-mqtt/blob/master/mqtt_as/mqtt_as.py#L361). I'm a bit puzzled: what is the value of republishing on the same connection? I.e., given that the connection is TCP-based, what kind of error does this try to overcome. It seems to me that if no ACK is received on the connection then there's no point resending on the same connection.

[Feature Questions] Unsubscribe, recognize retained messages

Hello Peter,

I implemented the functionality for "unsubscribe" into my fork.
Additionaly I added the possibility to recognize retained publications and pass that information on to the "subs_cb" callback. Because of this, my repository is not a drop-in replacement anymore as it requires the "subs_cb" to accept 3 args (topic,msg,retained).

I'd be happy if you could take a look at the changes of the commit adding the functionality. I'm not sure if I missed something but it definitely works as expected.
I'm also writing, because my "experimental" fork is mentioned in your README and with these changes you might be thinking differently about how to mention my fork or if you want to mention it at all, given that the functionality and drop-in ability has changed.

Lock object has no attribute '__aexit__'

After several hours of testing mqtt_as connected to the broker, the program ended with a message:

  File "mqtt_as.py"`, line 568, in _handle_msg

AttributeError: 'Lock' object has no attribute __aexit__

mqtt_as.py line: 568

565:   async def _handle_msg(self):
566:        try:
567:            while self.isconnected():
568:                async with self.lock:
569:                    await self.wait_msg()  # Immediate return if no message
570:                await asyncio.sleep_ms(_DEFAULT_MS)  # Let other tasks get lock
571:
572:        except OSError:
573:           pass
574:        self._reconnect()  # Broker or WiFi fail.

file mqtt_as.py has Lock() class at the beginning

class Lock():
    def __init__(self):
        self._locked = False

    async def __aenter__(self):
        while True:

            if self._locked:
                await asyncio.sleep_ms(_DEFAULT_MS)
            else:
                self._locked = True
                break

    async def __aexit__(self, *args):
        self._locked = False
        await asyncio.sleep_ms(_DEFAULT_MS)

version mqtt_as.py : from commit ff6b5aa 2019-11-04 19:04:16 with the change described in #10 (comment)

Usage question: Parallel publications

I'm gone through the docs but I'm still unclear whether submitting publications in parallel using asyncio is recommended or not. I've tested it with uasyncio.gather, but I had to add qos=1 because some publications were missing. I did this because for web sockets, asyncio is considered a good solution (hence the typical example of running a dozen http requests with asyncio).

So the question is what is the recommended practice when needing to publish to a few topics at once: 1) submitting the tasks in parallel and doing uasyncio.gather or 2) awaiting on them serially (in case they could step on each other).

I tested and both work I just wanted to know which one is more performant given the implementation.

Error 'module' object has no attribute 'Lock'

Hello,

i tried to install and use the resilient mqtt library. As you described in your Guide, i installed uasyncio, uasyncio.synchro and uasyncio.queues Link and copied mqtt_as.py and config.py via mpfshell to the root directory. Unfortunately, im stuck with the error "'module' object has no attribute 'Lock'" when i call a test script. Looking at the code in mqtt_as, it seems this error comes from lock being not found in uasyncio. Can you help me with that?

Anyway, thank you for the effort to create a resilient mqtt lib and the great documentation. I can't wait to start working with the library.

Don't Init Wifi if already connected on ESP32

I noticed this line skips initializing WiFi if it's already active, but only for ESP8266. But I think this would be very helpful on ESP32 as well, if you first initialize WiFi for NTP, FTP, etc. setup, then setup MQTT in your final code. I experienced MQTT connect issues when it tries to immediately reconnect on an existing connection. Any reason this test should not be performed for all devices? I tried that and it works fine on ESP32. Thanks for mqtt_as!

mqtt_as: How to catch exceptions when subscribtion callback throws one?

I'm using the mqtt_as MQTT implementation and doing my first uasyncio steps with the provided tutorial - thanks for both a lot btw!

I ended up with a very simple what I'd call mainloop in the end of my main script:

async def main_loop():                                                          
    while True:                                                                                        
        await asyncio.sleep(0)

asyncio.run(main_loop())

However wen I now send MQTT messages to topics I previously subscribed to - which throw exceptions - I really don't know how to handle them. When purposely publishing a message where my subscription callback function would throw an exception, I get:

From my script I get:

INFO:core:MQTT topic subscription callback called
DEBUG:core:  MQTT topic:   X/Y/Z
DEBUG:core:  MQTT payload: b'foobar'
ERROR:core:MQTT (sub)topic unknown -> throw MqttTopicException)
Task exception wasn't retrieved     <<<<<------------------------
future: <Task> coro= <generator object '_handle_msg' at 7f5b569e2920>
Traceback (most recent call last):
  File "uasyncio/core.py", line 1, in run_until_complete
  File "mqtt_as/mqtt_as.py", line 546, in _handle_msg
  File "mqtt_as/mqtt_as.py", line 439, in wait_msg
  File "upyot/lib/Mqtt.py", line 364, in _cb_mqtt_sub
  File "upyot/lib/Mqtt.py", line 344, in _cb_mqtt_sub
MqttTopicException: UNKNOWN_TOPIC

Wrapping await asyncio.sleep(0) with try/except doesn't catch it either. So how to do?
Sorry if this is a trivial question, maybe this might be one also addressed in the tutorial, e.g. under "5. Exceptions"?

Example program unclean.py

With ref to my TinyPico ESP32 board:
When I control C to quit to the unclean.py example program to experiment such as changing the subscription topic, when I run the program again I always get:
Connection failed.
Wifi Internal Error

After a bit of juggling I found that including an 'except' for ctrl C in the try / except that runs the program to include a statement to make the wifi inactive solves this and the program will now always restart ok

My amended try except is shown below
try:
asyncio.run(main(client))

except KeyboardInterrupt:
print('control C')
client._sta_if.active(False)

finally: # Prevent LmacRxBlk:1 errors.
client.close()
asyncio.new_event_loop()

So I mention this as an issue in case it helps others who may be loosing hair because of continually having to unplug their board just to re-run the program after some small experiment. And also I mention it in case this is addition is deemed to be a silly thing to do. (yes I'm a beginner), and if its OK then maybe it could be incorporated into the example code.

Whilst I'm here, I will say a big thanks for this program mqtt program, and also for the ESP8266 wifi bridge program (and image) for the rpi-pico. It works just great. Sad I did not see this program before buying an Adafruit airlift board to try to do what I can now do with on of my old Wemos Esp8266 boards. PS an ESP8266 image for the mqtt-as (as you do for the bridge) would help a lot for dummies like me who have not had much luck in creating there own frozen byte code image.

mqtt_as.py not recovering when starting while broker on host is down.

Hi Peter,

When I start with terminating my mosquitto broker and then start initialising the mqtt client an exception is raised:

Traceback (most recent call last):
File "uasyncio/core.py", line 1, in run_until_complete
File "mqtt_as.py", line 519, in connect
File "mqtt_as.py", line 248, in _connect
File "mqtt_as.py", line 188, in _as_write
OSError: [Errno 104] ECONNRESET

This seems to bubble up to the task itself, so it does not automatically recover.
Would you consider this a bug and is it worth fixing here?

I can imagine this to be a scenario where you've explicitly chosen not to recover, but for my scenario it's something I should expect and I'll have to find a fix to recover from this.

regards,
Chris

Pycom WiFi connection

When using your library in a Pycom board, the WiFi connection process and methods are different from a regular ESP32, so it returns an exception. I could modify your code in order to detect hardware and use the appropriate connection method, but I think it would be nice to have it in the original library.

Subscriptions lost after network re-connect

I have recently started using the mqtt library.
I periodically experience that the wifi re-connects. I dont know why this happens.
After re-connection the publishing continues without problems, but it seems all subscriptions are lost. The device becomes unresponsive to messages published by server.
I have copied in a piece of log below where the reset occurs.

Is there a way to avoid this or make automatic re-subscriptions?

Log extract:

RAM free 80016 alloc 31152
{"unitId":"kumme_stue","sensor":"ms-2","pin":32,"value":163.45 }
{"unitId":"kumme_stue","sensor":"ms-1","pin":35,"value":2024.78 }
loop 49
RAM free 79296 alloc 31872
loop 50
RAM free 79296 alloc 31872
loop 51
I (783748) wifi:state: run -> init (0)
I (783748) wifi:pm stop, total sleep time: 673555049 us / 773555826 us

I (783748) wifi:new:<6,0>, old:<6,0>, ap:<255,255>, sta:<6,0>, prof:1
I (783758) wifi: STA_DISCONNECTED, reason:8
I (784878) wifi:new:<6,0>, old:<6,0>, ap:<255,255>, sta:<6,0>, prof:1
I (784878) wifi:state: init -> auth (b0)
I (784878) wifi:state: auth -> assoc (0)
I (784888) wifi:state: assoc -> run (10)
I (784918) wifi:connected with Grootsvej10 IV, aid = 2, channel 6, BW20, bssid = f4:17:b8:81:f1:b2
I (784918) wifi:security: WPA2-PSK, phy: bgn, rssi: -89
I (784928) wifi:pm start, type: 1

I (784928) network: CONNECTED
I (784948) wifi:AP's beacon interval = 102400 us, DTIM period = 1
I (785828) tcpip_adapter: sta ip: 192.168.1.222, mask: 255.255.255.0, gw: 192.168.1.254
I (785828) network: GOT_IP
Checking WiFi integrity.
Got reliable connection
Connecting to broker.
Connected to broker.
Reconnect OK!
MQTT Client connected
loop 52
loop 53
RAM free 79968 alloc 31200
{"unitId":"kumme_stue","sensor":"ms-2","pin":32,"value":161.8718 }
{"unitId":"kumme_stue","sensor":"ms-1","pin":35,"value":2010.1 }
loop 54
RAM free 79952 alloc 31216
loop 55
{"unitId":"kumme_stue","sensor":"ms-2","pin":32,"value":152.5946 }
{"unitId":"kumme_stue","sensor":"ms-1","pin":35,"value":2003.66 }
RAM free 79952 alloc 31216
loop 56
loop 57

Implementing a WDT timer

Hi,

after extensive reading, I've implemented your MQTT client with success, thanks so much.

However after few days my ESP8266 stopped working. I don't think it's because of wifi outage but I suspect, since the microchips are close to a pump inverter (which produce quite a powerfull magnetic field) that that could be the reason.

I was wondering how I could implement machine.wdt() with your code in presence of asyncio coroutines which I'm quite new to.

Should I just declare as in the docs :

from machine import WDT
wdt = WDT(timeout=300000) 

and then in the main loop simply add a wdt.feed() like the following ?

while True:
  try:
      ntptime.settime()
      rtc=machine.RTC()
  except:
      continue

  print('publishing temps')
  # If WiFi is down the following will pause for the duration.
  json_data = ujson.dumps(getTemps())
  await client.publish('VMS_TEMPS_198280198239', json_data , qos = 1)
  wdt.feed()
  await asyncio.sleep(300)

I thank you in advance,

Sébastien

await client.connect() error in documented tutorial

https://www.raspberrypi.org/forums/viewtopic.php?f=32&t=243760
refers.

"async def main(client):
await client.connect()
n=0
"
fails with syntax error:-

"import asyncmqtttest.py
Traceback (most recent call last):
File "", line 1, in
File "asyncmqtttest.py", line 39
SyntaxError: invalid syntax"

Not good enough to post unvalidated code or faulty undocumented implementation.

Starting a MQTTClient with an already working wifi connection fails with Wifi internal error.

I discovered this while having the webrepl start in my boot.py, earlier than the MQTTClient start.
Running on micropython 1.16-135 on ESP32.

mac --> 24:0a:c4:ec:ab:1c
connecting to network...
('192.168.0.28', '255.255.255.0', '192.168.0.1', '12.123.123.123')
WebREPL daemon started on ws://192.168.0.28:8266
Started webrepl in normal mode
E (4600131) wifi:sta is connecting, return error
WebREPL daemon started on ws://192.168.0.28:8266
Started webrepl in normal mode
Traceback (most recent call last):
  File "main.py", line 69, in <module>
  File "uasyncio/core.py", line 1, in run
  File "uasyncio/core.py", line 1, in run_until_complete
  File "uasyncio/core.py", line 1, in run_until_complete
  File "main.py", line 50, in main
  File "mqtt_as.py", line 485, in wifi_connect
OSError: Wifi Internal Error
MicroPython v1.16-135-gaecb697c7-dirty on 2021-07-28; ESP32 module with ESP32
Type "help()" for more information.
>>> 

If I override the default wifi_connect, it works fine:

class MyMQTTClient(MQTTClient):

    async def wifi_connect(self):
        if not self._sta_if.isconnected():
            return await super().wifi_connect()

mqtt_as: PUBACKs of messages sent with qos=1 not handled and stacking up, resulting in 1) resends, then 2) no further messages being sent and 3) eventually resulting in mem exhaustion

I noticed when sending out messages via mqtt_as's publish() with qos=1 that at some point messages get resent to the broker.
Adding debug output to "mqtt_as" shows that at some point PUBACKs are not received/handled in time anymore, resulting in messages being resent.

At some even further point, no messages make it through to the broker at all anymore.

With below test case I send out 10 messages (in 2 sets of 5) every 15 seconds (triggered by ESP32's timer). After each sets uasyncio.sleep(4) is called.
Besides those 2x4 seconds there's also the mainloop calling uasyncio.sleep(10), so there should be plenty of scheduling time for mqtt_as to handle the PUBACKs (right?).

Something seems to stack up, though, as eventually the Guru starts to medidate with an MemoryError or it results in a crash caused by an assert() dropping:

assertion "mp_obj_is_small_int(args[2])" failed: file "../../extmod/moduasyncio.c", line 103, function: task_queue_push_sorted
abort() was called at PC 0x4016b75f on core 0

What is happening, what am I doing wrong? I was fearing it might be intrpt() being implicitly called via mp_sched_schedule() internally, but I'm kind of lost here.

The problem does not seem to occur with qos=0.

Micropython firmware version is v1.14 for ESP32 (SDK IDF v4, no psRAM).

Test code:

from machine import Pin, Timer
import network
import uasyncio
import utime
from mqtt_as.mqtt_as import MQTTClient, config

mqtt = None
i = 0

async def asyncfun():
    print("|| Scheduling 1st set of 5 messages..", utime.time())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    await uasyncio.sleep(4)
    print("|| Expected PUBACKs for 1st set of 5 messages to be handled by now", utime.time())
    print("|| Scheduling 2nd set of 5 messages..", utime.time())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    uasyncio.create_task(mqtt_pub())
    await uasyncio.sleep(4)
    print("|| Expected PUBACKs for 2nd set of 5 messages to be handled by now", utime.time())
    #print("asyncfun: done")

def intrpt(_=None):
    print("+++ TMR INT TRIGGERED", utime.time())
    print("|| Starting asyncfun()", utime.time())
    uasyncio.run(asyncfun())
    print("|| Finished asyncfun()", utime.time())

async def main_loop():
    while True:
        print("-- MAIN LOOP", utime.time())
        await uasyncio.sleep(10)

async def mqtt_pub():
    global mqtt
    global i
    i += 1
    print("-> PUBLISH", utime.time())
    await mqtt.publish("foo/bar", "Round {}".format(i), qos=1)
    print("-> /PUBLISH", utime.time())

async def mqtt_conn_cb(clnt):
    print("++ MQTT CONN CALLBACK", utime.time())
    print("Starting timer causing interrupts every 15secs..")
    Timer(0).init(mode=Timer.PERIODIC, period=15000, callback=intrpt)

async def mqtt_subs_cb(tpc, msg, rtnd):
    print("MQTT SUB CALLBACK!")

async def mqtt_connect():
    global mqtt
    config['ssid'] = 'foobar'
    config['wifi_pw'] = 'foobarfoobar'
    config['server'] = 'test.mosquitto.org'
    config['port'] = 1883
    config['client_id'] = 'fnord'
    config['clean'] = True
    config['response_time'] = 10
    config['subs_cb'] = mqtt_subs_cb
    config['connect_coro'] = mqtt_conn_cb

    print("++ CONNECTING TO WIFI/MQTT", utime.time())
    mqtt = MQTTClient(config)
    await mqtt.connect()

uasyncio.run(mqtt_connect())

uasyncio.run(main_loop())

Diff I applied to mqtt_as.py for debugging purposes:

diff --git a/mqtt_as/mqtt_as.py b/mqtt_as/mqtt_as.py
index ddf8d17..cf4abee 100644
--- a/mqtt_as/mqtt_as.py
+++ b/mqtt_as/mqtt_as.py
@@ -14,7 +14,7 @@ from ubinascii import hexlify
 import uasyncio as asyncio
 
 gc.collect()
-from utime import ticks_ms, ticks_diff
+from utime import ticks_ms, ticks_diff, time
 from uerrno import EINPROGRESS, ETIMEDOUT
 
 gc.collect()
@@ -341,14 +341,18 @@ class MQTT_base:
         count = 0
         while 1:  # Await PUBACK, republish on timeout
             if await self._await_pid(pid):
+                print(">#>#>#> RECEIVED PUBACK FOR PID {} (tpc: {} | msg: {})".format(pid, topic, msg), time())
                 return
+            print(">#>#>#> NO PUBACK IN TIME FOR PID {} (tpc: {} | msg: {})".format(pid, topic, msg), time())
             # No match
             if count >= self._max_repubs or not self.isconnected():
                 raise OSError(-1)  # Subclass to re-publish with new PID
+            print(">#>#>#> GOING TO *RE*PUBLISH PID {} (tpc: {} | msg: {})".format(pid, topic, msg), time())
             async with self.lock:
                 await self._publish(topic, msg, retain, qos, dup=1, pid=pid)  # Add pid
             count += 1
             self.REPUB_COUNT += 1
+            print(">#>#>#> *RE*PUBLISHED PID {} (ATTEMPT: {} (TOTAL: {})) (tpc: {} | msg: {})".format(pid, count, self.REPUB_COUNT, topic, msg), time())
 
     async def _publish(self, topic, msg, retain, qos, dup, pid):
         pkt = bytearray(b"\x30\0\0\0")

Log of example run:

>>> MicroPython v1.14 on 2021-02-02; ESP32 module with ESP32
Type "help()" for more information.
>>> 
>>> import test
++ CONNECTING TO WIFI/MQTT 3
++ MQTT CONN CALLBACK 12
Starting timer causing interrupts every 15secs..
-- MAIN LOOP 12
-- MAIN LOOP 22
+++ TMR INT TRIGGERED 27
|| Starting asyncfun() 27
|| Scheduling 1st set of 5 messages.. 27
-> PUBLISH 27
-> PUBLISH 27
-> PUBLISH 27
-> PUBLISH 27
-> PUBLISH 27
>#>#>#> RECEIVED PUBACK FOR PID 1 (tpc: foo/bar | msg: Round 1) 28
-> /PUBLISH 28
>#>#>#> RECEIVED PUBACK FOR PID 2 (tpc: foo/bar | msg: Round 2) 28
-> /PUBLISH 28
>#>#>#> RECEIVED PUBACK FOR PID 3 (tpc: foo/bar | msg: Round 3) 29
-> /PUBLISH 29
>#>#>#> RECEIVED PUBACK FOR PID 4 (tpc: foo/bar | msg: Round 4) 29
-> /PUBLISH 29
>#>#>#> RECEIVED PUBACK FOR PID 5 (tpc: foo/bar | msg: Round 5) 29
-> /PUBLISH 29
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 31
|| Scheduling 2nd set of 5 messages.. 31
-> PUBLISH 31
-> PUBLISH 31
-> PUBLISH 31
-> PUBLISH 31
-> PUBLISH 31
>#>#>#> RECEIVED PUBACK FOR PID 6 (tpc: foo/bar | msg: Round 6) 31
-> /PUBLISH 31
>#>#>#> RECEIVED PUBACK FOR PID 7 (tpc: foo/bar | msg: Round 7) 31
-> /PUBLISH 31
-- MAIN LOOP 32
>#>#>#> RECEIVED PUBACK FOR PID 8 (tpc: foo/bar | msg: Round 8) 32
-> /PUBLISH 32
>#>#>#> RECEIVED PUBACK FOR PID 9 (tpc: foo/bar | msg: Round 9) 32
-> /PUBLISH 32
>#>#>#> RECEIVED PUBACK FOR PID 10 (tpc: foo/bar | msg: Round 10) 32
-> /PUBLISH 32
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 35
|| Finished asyncfun() 35
+++ TMR INT TRIGGERED 42
|| Starting asyncfun() 42
|| Scheduling 1st set of 5 messages.. 42
-> PUBLISH 42
-> PUBLISH 42
-> PUBLISH 42
-> PUBLISH 42
-> PUBLISH 42
-- MAIN LOOP 42
>#>#>#> RECEIVED PUBACK FOR PID 11 (tpc: foo/bar | msg: Round 11) 42
-> /PUBLISH 42
>#>#>#> RECEIVED PUBACK FOR PID 12 (tpc: foo/bar | msg: Round 12) 42
-> /PUBLISH 42
>#>#>#> RECEIVED PUBACK FOR PID 13 (tpc: foo/bar | msg: Round 13) 42
-> /PUBLISH 42
>#>#>#> RECEIVED PUBACK FOR PID 14 (tpc: foo/bar | msg: Round 14) 43
-> /PUBLISH 43
>#>#>#> RECEIVED PUBACK FOR PID 15 (tpc: foo/bar | msg: Round 15) 43
-> /PUBLISH 43
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 46
|| Scheduling 2nd set of 5 messages.. 46
-> PUBLISH 46
-> PUBLISH 46
-> PUBLISH 46
-> PUBLISH 46
-> PUBLISH 46
>#>#>#> RECEIVED PUBACK FOR PID 16 (tpc: foo/bar | msg: Round 16) 46
-> /PUBLISH 46
>#>#>#> RECEIVED PUBACK FOR PID 17 (tpc: foo/bar | msg: Round 17) 46
-> /PUBLISH 46
>#>#>#> RECEIVED PUBACK FOR PID 18 (tpc: foo/bar | msg: Round 18) 46
-> /PUBLISH 46
>#>#>#> RECEIVED PUBACK FOR PID 19 (tpc: foo/bar | msg: Round 19) 47
-> /PUBLISH 47
>#>#>#> RECEIVED PUBACK FOR PID 20 (tpc: foo/bar | msg: Round 20) 47
-> /PUBLISH 47
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 50
|| Finished asyncfun() 50
-- MAIN LOOP 52
+++ TMR INT TRIGGERED 57
|| Starting asyncfun() 57
|| Scheduling 1st set of 5 messages.. 57
-> PUBLISH 57
-> PUBLISH 57
-> PUBLISH 57
-> PUBLISH 57
-> PUBLISH 57
>#>#>#> RECEIVED PUBACK FOR PID 21 (tpc: foo/bar | msg: Round 21) 57
-> /PUBLISH 57
>#>#>#> RECEIVED PUBACK FOR PID 22 (tpc: foo/bar | msg: Round 22) 57
-> /PUBLISH 57
>#>#>#> RECEIVED PUBACK FOR PID 23 (tpc: foo/bar | msg: Round 23) 57
-> /PUBLISH 57
>#>#>#> RECEIVED PUBACK FOR PID 24 (tpc: foo/bar | msg: Round 24) 57
-> /PUBLISH 57
>#>#>#> RECEIVED PUBACK FOR PID 25 (tpc: foo/bar | msg: Round 25) 58
-> /PUBLISH 58
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 61
|| Scheduling 2nd set of 5 messages.. 61
-> PUBLISH 61
-> PUBLISH 61
-> PUBLISH 61
-> PUBLISH 61
-> PUBLISH 61
-- MAIN LOOP 62
>#>#>#> RECEIVED PUBACK FOR PID 26 (tpc: foo/bar | msg: Round 26) 62
-> /PUBLISH 62
>#>#>#> RECEIVED PUBACK FOR PID 27 (tpc: foo/bar | msg: Round 27) 62
-> /PUBLISH 62
>#>#>#> RECEIVED PUBACK FOR PID 28 (tpc: foo/bar | msg: Round 28) 62
-> /PUBLISH 62
>#>#>#> RECEIVED PUBACK FOR PID 29 (tpc: foo/bar | msg: Round 29) 62
-> /PUBLISH 62
>#>#>#> RECEIVED PUBACK FOR PID 30 (tpc: foo/bar | msg: Round 30) 62
-> /PUBLISH 62
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 65
|| Finished asyncfun() 65
+++ TMR INT TRIGGERED 72
|| Starting asyncfun() 72
|| Scheduling 1st set of 5 messages.. 72
-> PUBLISH 72
-> PUBLISH 72
-> PUBLISH 72
-> PUBLISH 72
-> PUBLISH 72
-- MAIN LOOP 72
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 76
|| Scheduling 2nd set of 5 messages.. 76
-> PUBLISH 76
-> PUBLISH 76
-> PUBLISH 76
-> PUBLISH 76
-> PUBLISH 76
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 80
|| Finished asyncfun() 80
-- MAIN LOOP 82
+++ TMR INT TRIGGERED 87
|| Starting asyncfun() 87
|| Scheduling 1st set of 5 messages.. 87
-> PUBLISH 87
-> PUBLISH 87
-> PUBLISH 87
-> PUBLISH 87
-> PUBLISH 87
>#>#>#> NO PUBACK IN TIME FOR PID 31 (tpc: foo/bar | msg: Round 31) 90
>#>#>#> GOING TO *RE*PUBLISH PID 31 (tpc: foo/bar | msg: Round 31) 90
>#>#>#> *RE*PUBLISHED PID 31 (ATTEMPT: 1 (TOTAL: 1)) (tpc: foo/bar | msg: Round 31) 90
>#>#>#> NO PUBACK IN TIME FOR PID 32 (tpc: foo/bar | msg: Round 32) 90
>#>#>#> GOING TO *RE*PUBLISH PID 32 (tpc: foo/bar | msg: Round 32) 90
>#>#>#> NO PUBACK IN TIME FOR PID 33 (tpc: foo/bar | msg: Round 33) 90
>#>#>#> GOING TO *RE*PUBLISH PID 33 (tpc: foo/bar | msg: Round 33) 90
>#>#>#> *RE*PUBLISHED PID 32 (ATTEMPT: 1 (TOTAL: 2)) (tpc: foo/bar | msg: Round 32) 90
>#>#>#> NO PUBACK IN TIME FOR PID 34 (tpc: foo/bar | msg: Round 34) 90
>#>#>#> GOING TO *RE*PUBLISH PID 34 (tpc: foo/bar | msg: Round 34) 90
>#>#>#> NO PUBACK IN TIME FOR PID 37 (tpc: foo/bar | msg: Round 37) 90
>#>#>#> GOING TO *RE*PUBLISH PID 37 (tpc: foo/bar | msg: Round 37) 90
>#>#>#> *RE*PUBLISHED PID 33 (ATTEMPT: 1 (TOTAL: 3)) (tpc: foo/bar | msg: Round 33) 90
>#>#>#> NO PUBACK IN TIME FOR PID 35 (tpc: foo/bar | msg: Round 35) 90
>#>#>#> GOING TO *RE*PUBLISH PID 35 (tpc: foo/bar | msg: Round 35) 90
>#>#>#> NO PUBACK IN TIME FOR PID 38 (tpc: foo/bar | msg: Round 38) 90
>#>#>#> GOING TO *RE*PUBLISH PID 38 (tpc: foo/bar | msg: Round 38) 90
>#>#>#> NO PUBACK IN TIME FOR PID 36 (tpc: foo/bar | msg: Round 36) 90
>#>#>#> GOING TO *RE*PUBLISH PID 36 (tpc: foo/bar | msg: Round 36) 90
>#>#>#> NO PUBACK IN TIME FOR PID 39 (tpc: foo/bar | msg: Round 39) 90
>#>#>#> GOING TO *RE*PUBLISH PID 39 (tpc: foo/bar | msg: Round 39) 90
>#>#>#> *RE*PUBLISHED PID 34 (ATTEMPT: 1 (TOTAL: 4)) (tpc: foo/bar | msg: Round 34) 90
>#>#>#> NO PUBACK IN TIME FOR PID 40 (tpc: foo/bar | msg: Round 40) 90
>#>#>#> GOING TO *RE*PUBLISH PID 40 (tpc: foo/bar | msg: Round 40) 90
>#>#>#> *RE*PUBLISHED PID 37 (ATTEMPT: 1 (TOTAL: 5)) (tpc: foo/bar | msg: Round 37) 90
>#>#>#> *RE*PUBLISHED PID 35 (ATTEMPT: 1 (TOTAL: 6)) (tpc: foo/bar | msg: Round 35) 90
>#>#>#> *RE*PUBLISHED PID 38 (ATTEMPT: 1 (TOTAL: 7)) (tpc: foo/bar | msg: Round 38) 90
>#>#>#> *RE*PUBLISHED PID 36 (ATTEMPT: 1 (TOTAL: 8)) (tpc: foo/bar | msg: Round 36) 90
>#>#>#> *RE*PUBLISHED PID 39 (ATTEMPT: 1 (TOTAL: 9)) (tpc: foo/bar | msg: Round 39) 90
>#>#>#> *RE*PUBLISHED PID 40 (ATTEMPT: 1 (TOTAL: 10)) (tpc: foo/bar | msg: Round 40) 90
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 91
|| Scheduling 2nd set of 5 messages.. 91
-> PUBLISH 91
-> PUBLISH 91
-> PUBLISH 91
-> PUBLISH 91
-> PUBLISH 91
-- MAIN LOOP 92
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 95
|| Finished asyncfun() 95
>#>#>#> NO PUBACK IN TIME FOR PID 41 (tpc: foo/bar | msg: Round 41) 97
>#>#>#> GOING TO *RE*PUBLISH PID 41 (tpc: foo/bar | msg: Round 41) 97
>#>#>#> *RE*PUBLISHED PID 41 (ATTEMPT: 1 (TOTAL: 11)) (tpc: foo/bar | msg: Round 41) 97
>#>#>#> NO PUBACK IN TIME FOR PID 43 (tpc: foo/bar | msg: Round 43) 97
>#>#>#> GOING TO *RE*PUBLISH PID 43 (tpc: foo/bar | msg: Round 43) 97
>#>#>#> NO PUBACK IN TIME FOR PID 44 (tpc: foo/bar | msg: Round 44) 97
>#>#>#> GOING TO *RE*PUBLISH PID 44 (tpc: foo/bar | msg: Round 44) 97
>#>#>#> NO PUBACK IN TIME FOR PID 42 (tpc: foo/bar | msg: Round 42) 97
>#>#>#> GOING TO *RE*PUBLISH PID 42 (tpc: foo/bar | msg: Round 42) 97
>#>#>#> NO PUBACK IN TIME FOR PID 45 (tpc: foo/bar | msg: Round 45) 97
>#>#>#> GOING TO *RE*PUBLISH PID 45 (tpc: foo/bar | msg: Round 45) 97
>#>#>#> *RE*PUBLISHED PID 43 (ATTEMPT: 1 (TOTAL: 12)) (tpc: foo/bar | msg: Round 43) 97
>#>#>#> *RE*PUBLISHED PID 44 (ATTEMPT: 1 (TOTAL: 13)) (tpc: foo/bar | msg: Round 44) 97
>#>#>#> *RE*PUBLISHED PID 42 (ATTEMPT: 1 (TOTAL: 14)) (tpc: foo/bar | msg: Round 42) 97
>#>#>#> *RE*PUBLISHED PID 45 (ATTEMPT: 1 (TOTAL: 15)) (tpc: foo/bar | msg: Round 45) 97
>#>#>#> NO PUBACK IN TIME FOR PID 31 (tpc: foo/bar | msg: Round 31) 100
>#>#>#> GOING TO *RE*PUBLISH PID 31 (tpc: foo/bar | msg: Round 31) 100
>#>#>#> *RE*PUBLISHED PID 31 (ATTEMPT: 2 (TOTAL: 16)) (tpc: foo/bar | msg: Round 31) 100
>#>#>#> NO PUBACK IN TIME FOR PID 32 (tpc: foo/bar | msg: Round 32) 100
>#>#>#> GOING TO *RE*PUBLISH PID 32 (tpc: foo/bar | msg: Round 32) 100
>#>#>#> *RE*PUBLISHED PID 32 (ATTEMPT: 2 (TOTAL: 17)) (tpc: foo/bar | msg: Round 32) 100
>#>#>#> NO PUBACK IN TIME FOR PID 33 (tpc: foo/bar | msg: Round 33) 100
>#>#>#> GOING TO *RE*PUBLISH PID 33 (tpc: foo/bar | msg: Round 33) 100
>#>#>#> *RE*PUBLISHED PID 33 (ATTEMPT: 2 (TOTAL: 18)) (tpc: foo/bar | msg: Round 33) 100
>#>#>#> NO PUBACK IN TIME FOR PID 34 (tpc: foo/bar | msg: Round 34) 100
>#>#>#> GOING TO *RE*PUBLISH PID 34 (tpc: foo/bar | msg: Round 34) 100
>#>#>#> *RE*PUBLISHED PID 34 (ATTEMPT: 2 (TOTAL: 19)) (tpc: foo/bar | msg: Round 34) 100
>#>#>#> NO PUBACK IN TIME FOR PID 37 (tpc: foo/bar | msg: Round 37) 100
>#>#>#> GOING TO *RE*PUBLISH PID 37 (tpc: foo/bar | msg: Round 37) 100
>#>#>#> NO PUBACK IN TIME FOR PID 35 (tpc: foo/bar | msg: Round 35) 100
>#>#>#> GOING TO *RE*PUBLISH PID 35 (tpc: foo/bar | msg: Round 35) 100
>#>#>#> *RE*PUBLISHED PID 37 (ATTEMPT: 2 (TOTAL: 20)) (tpc: foo/bar | msg: Round 37) 100
>#>#>#> NO PUBACK IN TIME FOR PID 38 (tpc: foo/bar | msg: Round 38) 100
>#>#>#> GOING TO *RE*PUBLISH PID 38 (tpc: foo/bar | msg: Round 38) 100
>#>#>#> *RE*PUBLISHED PID 35 (ATTEMPT: 2 (TOTAL: 21)) (tpc: foo/bar | msg: Round 35) 100
>#>#>#> NO PUBACK IN TIME FOR PID 36 (tpc: foo/bar | msg: Round 36) 100
>#>#>#> GOING TO *RE*PUBLISH PID 36 (tpc: foo/bar | msg: Round 36) 100
>#>#>#> *RE*PUBLISHED PID 38 (ATTEMPT: 2 (TOTAL: 22)) (tpc: foo/bar | msg: Round 38) 100
>#>#>#> NO PUBACK IN TIME FOR PID 39 (tpc: foo/bar | msg: Round 39) 100
>#>#>#> GOING TO *RE*PUBLISH PID 39 (tpc: foo/bar | msg: Round 39) 100
>#>#>#> *RE*PUBLISHED PID 36 (ATTEMPT: 2 (TOTAL: 23)) (tpc: foo/bar | msg: Round 36) 101
>#>#>#> NO PUBACK IN TIME FOR PID 40 (tpc: foo/bar | msg: Round 40) 101
>#>#>#> GOING TO *RE*PUBLISH PID 40 (tpc: foo/bar | msg: Round 40) 101
>#>#>#> *RE*PUBLISHED PID 39 (ATTEMPT: 2 (TOTAL: 24)) (tpc: foo/bar | msg: Round 39) 101
>#>#>#> *RE*PUBLISHED PID 40 (ATTEMPT: 2 (TOTAL: 25)) (tpc: foo/bar | msg: Round 40) 101
>#>#>#> NO PUBACK IN TIME FOR PID 46 (tpc: foo/bar | msg: Round 46) 101
>#>#>#> GOING TO *RE*PUBLISH PID 46 (tpc: foo/bar | msg: Round 46) 101
>#>#>#> *RE*PUBLISHED PID 46 (ATTEMPT: 1 (TOTAL: 26)) (tpc: foo/bar | msg: Round 46) 101
>#>#>#> NO PUBACK IN TIME FOR PID 49 (tpc: foo/bar | msg: Round 49) 101
>#>#>#> GOING TO *RE*PUBLISH PID 49 (tpc: foo/bar | msg: Round 49) 101
>#>#>#> NO PUBACK IN TIME FOR PID 47 (tpc: foo/bar | msg: Round 47) 101
>#>#>#> GOING TO *RE*PUBLISH PID 47 (tpc: foo/bar | msg: Round 47) 101
>#>#>#> NO PUBACK IN TIME FOR PID 50 (tpc: foo/bar | msg: Round 50) 101
>#>#>#> GOING TO *RE*PUBLISH PID 50 (tpc: foo/bar | msg: Round 50) 101
>#>#>#> NO PUBACK IN TIME FOR PID 48 (tpc: foo/bar | msg: Round 48) 101
>#>#>#> GOING TO *RE*PUBLISH PID 48 (tpc: foo/bar | msg: Round 48) 101
>#>#>#> *RE*PUBLISHED PID 49 (ATTEMPT: 1 (TOTAL: 27)) (tpc: foo/bar | msg: Round 49) 101
>#>#>#> *RE*PUBLISHED PID 47 (ATTEMPT: 1 (TOTAL: 28)) (tpc: foo/bar | msg: Round 47) 101
>#>#>#> *RE*PUBLISHED PID 50 (ATTEMPT: 1 (TOTAL: 29)) (tpc: foo/bar | msg: Round 50) 101
>#>#>#> *RE*PUBLISHED PID 48 (ATTEMPT: 1 (TOTAL: 30)) (tpc: foo/bar | msg: Round 48) 101
+++ TMR INT TRIGGERED 102
|| Starting asyncfun() 102
|| Scheduling 1st set of 5 messages.. 102
-> PUBLISH 102
-> PUBLISH 102
-> PUBLISH 102
-> PUBLISH 102
-> PUBLISH 102
-- MAIN LOOP 102
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 106
|| Scheduling 2nd set of 5 messages.. 106
-> PUBLISH 106
-> PUBLISH 106
-> PUBLISH 106
-> PUBLISH 106
-> PUBLISH 106
>#>#>#> NO PUBACK IN TIME FOR PID 41 (tpc: foo/bar | msg: Round 41) 107
>#>#>#> GOING TO *RE*PUBLISH PID 41 (tpc: foo/bar | msg: Round 41) 107
>#>#>#> *RE*PUBLISHED PID 41 (ATTEMPT: 2 (TOTAL: 31)) (tpc: foo/bar | msg: Round 41) 107
>#>#>#> NO PUBACK IN TIME FOR PID 43 (tpc: foo/bar | msg: Round 43) 107
>#>#>#> GOING TO *RE*PUBLISH PID 43 (tpc: foo/bar | msg: Round 43) 107
>#>#>#> *RE*PUBLISHED PID 43 (ATTEMPT: 2 (TOTAL: 32)) (tpc: foo/bar | msg: Round 43) 107
>#>#>#> NO PUBACK IN TIME FOR PID 44 (tpc: foo/bar | msg: Round 44) 107
>#>#>#> GOING TO *RE*PUBLISH PID 44 (tpc: foo/bar | msg: Round 44) 107
>#>#>#> NO PUBACK IN TIME FOR PID 42 (tpc: foo/bar | msg: Round 42) 107
>#>#>#> GOING TO *RE*PUBLISH PID 42 (tpc: foo/bar | msg: Round 42) 107
>#>#>#> *RE*PUBLISHED PID 44 (ATTEMPT: 2 (TOTAL: 33)) (tpc: foo/bar | msg: Round 44) 107
>#>#>#> NO PUBACK IN TIME FOR PID 45 (tpc: foo/bar | msg: Round 45) 107
>#>#>#> GOING TO *RE*PUBLISH PID 45 (tpc: foo/bar | msg: Round 45) 107
>#>#>#> *RE*PUBLISHED PID 42 (ATTEMPT: 2 (TOTAL: 34)) (tpc: foo/bar | msg: Round 42) 107
>#>#>#> *RE*PUBLISHED PID 45 (ATTEMPT: 2 (TOTAL: 35)) (tpc: foo/bar | msg: Round 45) 107
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 110
|| Finished asyncfun() 110
>#>#>#> NO PUBACK IN TIME FOR PID 31 (tpc: foo/bar | msg: Round 31) 110
>#>#>#> GOING TO *RE*PUBLISH PID 31 (tpc: foo/bar | msg: Round 31) 110
>#>#>#> *RE*PUBLISHED PID 31 (ATTEMPT: 3 (TOTAL: 36)) (tpc: foo/bar | msg: Round 31) 110
>#>#>#> NO PUBACK IN TIME FOR PID 32 (tpc: foo/bar | msg: Round 32) 110
>#>#>#> GOING TO *RE*PUBLISH PID 32 (tpc: foo/bar | msg: Round 32) 110
>#>#>#> NO PUBACK IN TIME FOR PID 33 (tpc: foo/bar | msg: Round 33) 110
>#>#>#> GOING TO *RE*PUBLISH PID 33 (tpc: foo/bar | msg: Round 33) 110
>#>#>#> *RE*PUBLISHED PID 32 (ATTEMPT: 3 (TOTAL: 37)) (tpc: foo/bar | msg: Round 32) 110
>#>#>#> *RE*PUBLISHED PID 33 (ATTEMPT: 3 (TOTAL: 38)) (tpc: foo/bar | msg: Round 33) 110
>#>#>#> NO PUBACK IN TIME FOR PID 34 (tpc: foo/bar | msg: Round 34) 110
>#>#>#> GOING TO *RE*PUBLISH PID 34 (tpc: foo/bar | msg: Round 34) 110
>#>#>#> *RE*PUBLISHED PID 34 (ATTEMPT: 3 (TOTAL: 39)) (tpc: foo/bar | msg: Round 34) 110
>#>#>#> NO PUBACK IN TIME FOR PID 37 (tpc: foo/bar | msg: Round 37) 110
>#>#>#> GOING TO *RE*PUBLISH PID 37 (tpc: foo/bar | msg: Round 37) 110
>#>#>#> NO PUBACK IN TIME FOR PID 35 (tpc: foo/bar | msg: Round 35) 110
>#>#>#> GOING TO *RE*PUBLISH PID 35 (tpc: foo/bar | msg: Round 35) 110
>#>#>#> *RE*PUBLISHED PID 37 (ATTEMPT: 3 (TOTAL: 40)) (tpc: foo/bar | msg: Round 37) 111
>#>#>#> NO PUBACK IN TIME FOR PID 38 (tpc: foo/bar | msg: Round 38) 111
>#>#>#> GOING TO *RE*PUBLISH PID 38 (tpc: foo/bar | msg: Round 38) 111
>#>#>#> *RE*PUBLISHED PID 35 (ATTEMPT: 3 (TOTAL: 41)) (tpc: foo/bar | msg: Round 35) 111
>#>#>#> NO PUBACK IN TIME FOR PID 36 (tpc: foo/bar | msg: Round 36) 111
>#>#>#> GOING TO *RE*PUBLISH PID 36 (tpc: foo/bar | msg: Round 36) 111
>#>#>#> *RE*PUBLISHED PID 38 (ATTEMPT: 3 (TOTAL: 42)) (tpc: foo/bar | msg: Round 38) 111
>#>#>#> NO PUBACK IN TIME FOR PID 39 (tpc: foo/bar | msg: Round 39) 111
>#>#>#> GOING TO *RE*PUBLISH PID 39 (tpc: foo/bar | msg: Round 39) 111
>#>#>#> NO PUBACK IN TIME FOR PID 40 (tpc: foo/bar | msg: Round 40) 111
>#>#>#> GOING TO *RE*PUBLISH PID 40 (tpc: foo/bar | msg: Round 40) 111
>#>#>#> *RE*PUBLISHED PID 36 (ATTEMPT: 3 (TOTAL: 43)) (tpc: foo/bar | msg: Round 36) 111
>#>#>#> *RE*PUBLISHED PID 39 (ATTEMPT: 3 (TOTAL: 44)) (tpc: foo/bar | msg: Round 39) 111
>#>#>#> *RE*PUBLISHED PID 40 (ATTEMPT: 3 (TOTAL: 45)) (tpc: foo/bar | msg: Round 40) 111
>#>#>#> NO PUBACK IN TIME FOR PID 46 (tpc: foo/bar | msg: Round 46) 111
>#>#>#> GOING TO *RE*PUBLISH PID 46 (tpc: foo/bar | msg: Round 46) 111
>#>#>#> *RE*PUBLISHED PID 46 (ATTEMPT: 2 (TOTAL: 46)) (tpc: foo/bar | msg: Round 46) 111
>#>#>#> NO PUBACK IN TIME FOR PID 49 (tpc: foo/bar | msg: Round 49) 111
>#>#>#> GOING TO *RE*PUBLISH PID 49 (tpc: foo/bar | msg: Round 49) 111
>#>#>#> NO PUBACK IN TIME FOR PID 47 (tpc: foo/bar | msg: Round 47) 111
>#>#>#> GOING TO *RE*PUBLISH PID 47 (tpc: foo/bar | msg: Round 47) 111
>#>#>#> *RE*PUBLISHED PID 49 (ATTEMPT: 2 (TOTAL: 47)) (tpc: foo/bar | msg: Round 49) 111
>#>#>#> NO PUBACK IN TIME FOR PID 50 (tpc: foo/bar | msg: Round 50) 111
>#>#>#> GOING TO *RE*PUBLISH PID 50 (tpc: foo/bar | msg: Round 50) 111
>#>#>#> *RE*PUBLISHED PID 47 (ATTEMPT: 2 (TOTAL: 48)) (tpc: foo/bar | msg: Round 47) 111
>#>#>#> NO PUBACK IN TIME FOR PID 48 (tpc: foo/bar | msg: Round 48) 111
>#>#>#> GOING TO *RE*PUBLISH PID 48 (tpc: foo/bar | msg: Round 48) 111
>#>#>#> *RE*PUBLISHED PID 50 (ATTEMPT: 2 (TOTAL: 49)) (tpc: foo/bar | msg: Round 50) 111
>#>#>#> *RE*PUBLISHED PID 48 (ATTEMPT: 2 (TOTAL: 50)) (tpc: foo/bar | msg: Round 48) 111
-- MAIN LOOP 112
>#>#>#> NO PUBACK IN TIME FOR PID 51 (tpc: foo/bar | msg: Round 51) 112
>#>#>#> GOING TO *RE*PUBLISH PID 51 (tpc: foo/bar | msg: Round 51) 112
>#>#>#> NO PUBACK IN TIME FOR PID 52 (tpc: foo/bar | msg: Round 52) 112
>#>#>#> GOING TO *RE*PUBLISH PID 52 (tpc: foo/bar | msg: Round 52) 112
>#>#>#> *RE*PUBLISHED PID 51 (ATTEMPT: 1 (TOTAL: 51)) (tpc: foo/bar | msg: Round 51) 112
>#>#>#> NO PUBACK IN TIME FOR PID 53 (tpc: foo/bar | msg: Round 53) 112
>#>#>#> GOING TO *RE*PUBLISH PID 53 (tpc: foo/bar | msg: Round 53) 112
>#>#>#> *RE*PUBLISHED PID 52 (ATTEMPT: 1 (TOTAL: 52)) (tpc: foo/bar | msg: Round 52) 112
>#>#>#> NO PUBACK IN TIME FOR PID 54 (tpc: foo/bar | msg: Round 54) 112
>#>#>#> GOING TO *RE*PUBLISH PID 54 (tpc: foo/bar | msg: Round 54) 112
>#>#>#> NO PUBACK IN TIME FOR PID 55 (tpc: foo/bar | msg: Round 55) 112
>#>#>#> GOING TO *RE*PUBLISH PID 55 (tpc: foo/bar | msg: Round 55) 112
>#>#>#> *RE*PUBLISHED PID 53 (ATTEMPT: 1 (TOTAL: 53)) (tpc: foo/bar | msg: Round 53) 112
>#>#>#> *RE*PUBLISHED PID 54 (ATTEMPT: 1 (TOTAL: 54)) (tpc: foo/bar | msg: Round 54) 112
>#>#>#> *RE*PUBLISHED PID 55 (ATTEMPT: 1 (TOTAL: 55)) (tpc: foo/bar | msg: Round 55) 112
>#>#>#> NO PUBACK IN TIME FOR PID 56 (tpc: foo/bar | msg: Round 56) 116
>#>#>#> GOING TO *RE*PUBLISH PID 56 (tpc: foo/bar | msg: Round 56) 116
>#>#>#> NO PUBACK IN TIME FOR PID 57 (tpc: foo/bar | msg: Round 57) 116
>#>#>#> GOING TO *RE*PUBLISH PID 57 (tpc: foo/bar | msg: Round 57) 116
>#>#>#> NO PUBACK IN TIME FOR PID 58 (tpc: foo/bar | msg: Round 58) 116
>#>#>#> GOING TO *RE*PUBLISH PID 58 (tpc: foo/bar | msg: Round 58) 116
>#>#>#> *RE*PUBLISHED PID 56 (ATTEMPT: 1 (TOTAL: 56)) (tpc: foo/bar | msg: Round 56) 116
>#>#>#> NO PUBACK IN TIME FOR PID 59 (tpc: foo/bar | msg: Round 59) 116
>#>#>#> GOING TO *RE*PUBLISH PID 59 (tpc: foo/bar | msg: Round 59) 116
>#>#>#> NO PUBACK IN TIME FOR PID 60 (tpc: foo/bar | msg: Round 60) 116
>#>#>#> GOING TO *RE*PUBLISH PID 60 (tpc: foo/bar | msg: Round 60) 116
>#>#>#> *RE*PUBLISHED PID 57 (ATTEMPT: 1 (TOTAL: 57)) (tpc: foo/bar | msg: Round 57) 116
>#>#>#> *RE*PUBLISHED PID 58 (ATTEMPT: 1 (TOTAL: 58)) (tpc: foo/bar | msg: Round 58) 116
>#>#>#> *RE*PUBLISHED PID 59 (ATTEMPT: 1 (TOTAL: 59)) (tpc: foo/bar | msg: Round 59) 116
>#>#>#> *RE*PUBLISHED PID 60 (ATTEMPT: 1 (TOTAL: 60)) (tpc: foo/bar | msg: Round 60) 116
+++ TMR INT TRIGGERED 117
|| Starting asyncfun() 117
|| Scheduling 1st set of 5 messages.. 117
-> PUBLISH 117
-> PUBLISH 117
-> PUBLISH 117
-> PUBLISH 117
-> PUBLISH 117
>#>#>#> NO PUBACK IN TIME FOR PID 41 (tpc: foo/bar | msg: Round 41) 117
>#>#>#> GOING TO *RE*PUBLISH PID 41 (tpc: foo/bar | msg: Round 41) 117
>#>#>#> *RE*PUBLISHED PID 41 (ATTEMPT: 3 (TOTAL: 61)) (tpc: foo/bar | msg: Round 41) 117
>#>#>#> NO PUBACK IN TIME FOR PID 43 (tpc: foo/bar | msg: Round 43) 117
>#>#>#> GOING TO *RE*PUBLISH PID 43 (tpc: foo/bar | msg: Round 43) 117
>#>#>#> *RE*PUBLISHED PID 43 (ATTEMPT: 3 (TOTAL: 62)) (tpc: foo/bar | msg: Round 43) 117
>#>#>#> NO PUBACK IN TIME FOR PID 44 (tpc: foo/bar | msg: Round 44) 117
>#>#>#> GOING TO *RE*PUBLISH PID 44 (tpc: foo/bar | msg: Round 44) 117
>#>#>#> NO PUBACK IN TIME FOR PID 42 (tpc: foo/bar | msg: Round 42) 117
>#>#>#> GOING TO *RE*PUBLISH PID 42 (tpc: foo/bar | msg: Round 42) 117
>#>#>#> *RE*PUBLISHED PID 44 (ATTEMPT: 3 (TOTAL: 63)) (tpc: foo/bar | msg: Round 44) 117
>#>#>#> NO PUBACK IN TIME FOR PID 45 (tpc: foo/bar | msg: Round 45) 117
>#>#>#> GOING TO *RE*PUBLISH PID 45 (tpc: foo/bar | msg: Round 45) 117
>#>#>#> *RE*PUBLISHED PID 42 (ATTEMPT: 3 (TOTAL: 64)) (tpc: foo/bar | msg: Round 42) 117
>#>#>#> *RE*PUBLISHED PID 45 (ATTEMPT: 3 (TOTAL: 65)) (tpc: foo/bar | msg: Round 45) 117
>#>#>#> NO PUBACK IN TIME FOR PID 31 (tpc: foo/bar | msg: Round 31) 120
>#>#>#> GOING TO *RE*PUBLISH PID 31 (tpc: foo/bar | msg: Round 31) 120
>#>#>#> NO PUBACK IN TIME FOR PID 32 (tpc: foo/bar | msg: Round 32) 120
>#>#>#> GOING TO *RE*PUBLISH PID 32 (tpc: foo/bar | msg: Round 32) 120
>#>#>#> *RE*PUBLISHED PID 31 (ATTEMPT: 4 (TOTAL: 66)) (tpc: foo/bar | msg: Round 31) 120
>#>#>#> NO PUBACK IN TIME FOR PID 33 (tpc: foo/bar | msg: Round 33) 120
>#>#>#> GOING TO *RE*PUBLISH PID 33 (tpc: foo/bar | msg: Round 33) 120
>#>#>#> *RE*PUBLISHED PID 32 (ATTEMPT: 4 (TOTAL: 67)) (tpc: foo/bar | msg: Round 32) 120
>#>#>#> *RE*PUBLISHED PID 33 (ATTEMPT: 4 (TOTAL: 68)) (tpc: foo/bar | msg: Round 33) 120
>#>#>#> NO PUBACK IN TIME FOR PID 34 (tpc: foo/bar | msg: Round 34) 120
>#>#>#> GOING TO *RE*PUBLISH PID 34 (tpc: foo/bar | msg: Round 34) 120
>#>#>#> *RE*PUBLISHED PID 34 (ATTEMPT: 4 (TOTAL: 69)) (tpc: foo/bar | msg: Round 34) 121
>#>#>#> NO PUBACK IN TIME FOR PID 37 (tpc: foo/bar | msg: Round 37) 121
>#>#>#> GOING TO *RE*PUBLISH PID 37 (tpc: foo/bar | msg: Round 37) 121
>#>#>#> NO PUBACK IN TIME FOR PID 35 (tpc: foo/bar | msg: Round 35) 121
>#>#>#> GOING TO *RE*PUBLISH PID 35 (tpc: foo/bar | msg: Round 35) 121
>#>#>#> *RE*PUBLISHED PID 37 (ATTEMPT: 4 (TOTAL: 70)) (tpc: foo/bar | msg: Round 37) 121
>#>#>#> NO PUBACK IN TIME FOR PID 38 (tpc: foo/bar | msg: Round 38) 121
>#>#>#> GOING TO *RE*PUBLISH PID 38 (tpc: foo/bar | msg: Round 38) 121
>#>#>#> *RE*PUBLISHED PID 35 (ATTEMPT: 4 (TOTAL: 71)) (tpc: foo/bar | msg: Round 35) 121
>#>#>#> NO PUBACK IN TIME FOR PID 36 (tpc: foo/bar | msg: Round 36) 121
>#>#>#> GOING TO *RE*PUBLISH PID 36 (tpc: foo/bar | msg: Round 36) 121
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 121
|| Scheduling 2nd set of 5 messages.. 121
-> PUBLISH 121
-> PUBLISH 121
-> PUBLISH 121
-> PUBLISH 121
-> PUBLISH 121
>#>#>#> NO PUBACK IN TIME FOR PID 39 (tpc: foo/bar | msg: Round 39) 121
>#>#>#> GOING TO *RE*PUBLISH PID 39 (tpc: foo/bar | msg: Round 39) 121
>#>#>#> *RE*PUBLISHED PID 38 (ATTEMPT: 4 (TOTAL: 72)) (tpc: foo/bar | msg: Round 38) 121
>#>#>#> *RE*PUBLISHED PID 36 (ATTEMPT: 4 (TOTAL: 73)) (tpc: foo/bar | msg: Round 36) 121
>#>#>#> NO PUBACK IN TIME FOR PID 46 (tpc: foo/bar | msg: Round 46) 121
>#>#>#> GOING TO *RE*PUBLISH PID 46 (tpc: foo/bar | msg: Round 46) 121
>#>#>#> *RE*PUBLISHED PID 39 (ATTEMPT: 4 (TOTAL: 74)) (tpc: foo/bar | msg: Round 39) 121
>#>#>#> *RE*PUBLISHED PID 46 (ATTEMPT: 3 (TOTAL: 75)) (tpc: foo/bar | msg: Round 46) 121
>#>#>#> NO PUBACK IN TIME FOR PID 49 (tpc: foo/bar | msg: Round 49) 121
>#>#>#> GOING TO *RE*PUBLISH PID 49 (tpc: foo/bar | msg: Round 49) 121
>#>#>#> NO PUBACK IN TIME FOR PID 47 (tpc: foo/bar | msg: Round 47) 121
>#>#>#> GOING TO *RE*PUBLISH PID 47 (tpc: foo/bar | msg: Round 47) 121
>#>#>#> *RE*PUBLISHED PID 49 (ATTEMPT: 3 (TOTAL: 76)) (tpc: foo/bar | msg: Round 49) 121
>#>#>#> NO PUBACK IN TIME FOR PID 50 (tpc: foo/bar | msg: Round 50) 121
>#>#>#> GOING TO *RE*PUBLISH PID 50 (tpc: foo/bar | msg: Round 50) 121
>#>#>#> *RE*PUBLISHED PID 47 (ATTEMPT: 3 (TOTAL: 77)) (tpc: foo/bar | msg: Round 47) 121
>#>#>#> NO PUBACK IN TIME FOR PID 48 (tpc: foo/bar | msg: Round 48) 121
>#>#>#> GOING TO *RE*PUBLISH PID 48 (tpc: foo/bar | msg: Round 48) 121
>#>#>#> *RE*PUBLISHED PID 50 (ATTEMPT: 3 (TOTAL: 78)) (tpc: foo/bar | msg: Round 50) 121
>#>#>#> *RE*PUBLISHED PID 48 (ATTEMPT: 3 (TOTAL: 79)) (tpc: foo/bar | msg: Round 48) 122
-- MAIN LOOP 122
>#>#>#> NO PUBACK IN TIME FOR PID 51 (tpc: foo/bar | msg: Round 51) 122
>#>#>#> GOING TO *RE*PUBLISH PID 51 (tpc: foo/bar | msg: Round 51) 122
>#>#>#> *RE*PUBLISHED PID 51 (ATTEMPT: 2 (TOTAL: 80)) (tpc: foo/bar | msg: Round 51) 122
>#>#>#> NO PUBACK IN TIME FOR PID 52 (tpc: foo/bar | msg: Round 52) 122
>#>#>#> GOING TO *RE*PUBLISH PID 52 (tpc: foo/bar | msg: Round 52) 122
>#>#>#> *RE*PUBLISHED PID 52 (ATTEMPT: 2 (TOTAL: 81)) (tpc: foo/bar | msg: Round 52) 122
>#>#>#> NO PUBACK IN TIME FOR PID 53 (tpc: foo/bar | msg: Round 53) 122
>#>#>#> GOING TO *RE*PUBLISH PID 53 (tpc: foo/bar | msg: Round 53) 122
>#>#>#> NO PUBACK IN TIME FOR PID 54 (tpc: foo/bar | msg: Round 54) 122
>#>#>#> GOING TO *RE*PUBLISH PID 54 (tpc: foo/bar | msg: Round 54) 122
>#>#>#> NO PUBACK IN TIME FOR PID 55 (tpc: foo/bar | msg: Round 55) 122
>#>#>#> GOING TO *RE*PUBLISH PID 55 (tpc: foo/bar | msg: Round 55) 122
>#>#>#> *RE*PUBLISHED PID 53 (ATTEMPT: 2 (TOTAL: 82)) (tpc: foo/bar | msg: Round 53) 122
>#>#>#> NO PUBACK IN TIME FOR PID 58 (tpc: foo/bar | msg: Round 58) 122
>#>#>#> NO PUBACK IN TIME FOR PID 63 (tpc: foo/bar | msg: Round 63) 122
>#>#>#> NO PUBACK IN TIME FOR PID 42 (tpc: foo/bar | msg: Round 42) 122
>#>#>#> NO PUBACK IN TIME FOR PID 50 (tpc: foo/bar | msg: Round 50) 122
>#>#>#> NO PUBACK IN TIME FOR PID 66 (tpc: foo/bar | msg: Round 66) 122
>#>#>#> NO PUBACK IN TIME FOR PID 57 (tpc: foo/bar | msg: Round 57) 122
>#>#>#> NO PUBACK IN TIME FOR PID 61 (tpc: foo/bar | msg: Round 61) 122
>#>#>#> NO PUBACK IN TIME FOR PID 59 (tpc: foo/bar | msg: Round 59) 122
>#>#>#> NO PUBACK IN TIME FOR PID 32 (tpc: foo/bar | msg: Round 32) 122
>#>#>#> NO PUBACK IN TIME FOR PID 39 (tpc: foo/bar | msg: Round 39) 122
>#>#>#> NO PUBACK IN TIME FOR PID 64 (tpc: foo/bar | msg: Round 64) 122
>#>#>#> NO PUBACK IN TIME FOR PID 62 (tpc: foo/bar | msg: Round 62) 122
>#>#>#> NO PUBACK IN TIME FOR PID 34 (tpc: foo/bar | msg: Round 34) 122
>#>#>#> NO PUBACK IN TIME FOR PID 41 (tpc: foo/bar | msg: Round 41) 122
>#>#>#> NO PUBACK IN TIME FOR PID 37 (tpc: foo/bar | msg: Round 37) 122
>#>#>#> NO PUBACK IN TIME FOR PID 44 (tpc: foo/bar | msg: Round 44) 122
>#>#>#> NO PUBACK IN TIME FOR PID 53 (tpc: foo/bar | msg: Round 53) 122
>#>#>#> NO PUBACK IN TIME FOR PID 52 (tpc: foo/bar | msg: Round 52) 122
>#>#>#> NO PUBACK IN TIME FOR PID 43 (tpc: foo/bar | msg: Round 43) 122
>#>#>#> NO PUBACK IN TIME FOR PID 69 (tpc: foo/bar | msg: Round 69) 122
>#>#>#> NO PUBACK IN TIME FOR PID 47 (tpc: foo/bar | msg: Round 47) 122
>#>#>#> NO PUBACK IN TIME FOR PID 67 (tpc: foo/bar | msg: Round 67) 122
>#>#>#> NO PUBACK IN TIME FOR PID 36 (tpc: foo/bar | msg: Round 36) 122
>#>#>#> NO PUBACK IN TIME FOR PID 31 (tpc: foo/bar | msg: Round 31) 122
>#>#>#> NO PUBACK IN TIME FOR PID 48 (tpc: foo/bar | msg: Round 48) 122
>#>#>#> NO PUBACK IN TIME FOR PID 65 (tpc: foo/bar | msg: Round 65) 122
>#>#>#> NO PUBACK IN TIME FOR PID 56 (tpc: foo/bar | msg: Round 56) 123
>#>#>#> NO PUBACK IN TIME FOR PID 45 (tpc: foo/bar | msg: Round 45) 123
>#>#>#> NO PUBACK IN TIME FOR PID 33 (tpc: foo/bar | msg: Round 33) 123
>#>#>#> NO PUBACK IN TIME FOR PID 35 (tpc: foo/bar | msg: Round 35) 123
>#>#>#> NO PUBACK IN TIME FOR PID 70 (tpc: foo/bar | msg: Round 70) 123
>#>#>#> NO PUBACK IN TIME FOR PID 49 (tpc: foo/bar | msg: Round 49) 123
>#>#>#> NO PUBACK IN TIME FOR PID 46 (tpc: foo/bar | msg: Round 46) 123
>#>#>#> NO PUBACK IN TIME FOR PID 51 (tpc: foo/bar | msg: Round 51) 123
>#>#>#> NO PUBACK IN TIME FOR PID 38 (tpc: foo/bar | msg: Round 38) 123
>#>#>#> NO PUBACK IN TIME FOR PID 68 (tpc: foo/bar | msg: Round 68) 123
>#>#>#> NO PUBACK IN TIME FOR PID 60 (tpc: foo/bar | msg: Round 60) 123
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 125
|| Finished asyncfun() 125
>#>#>#> NO PUBACK IN TIME FOR PID 40 (tpc: foo/bar | msg: Round 40) 125
++ MQTT CONN CALLBACK 130
Starting timer causing interrupts every 15secs..
>#>#>#> RECEIVED PUBACK FOR PID 71 (tpc: foo/bar | msg: Round 53) 131
-> /PUBLISH 131
>#>#>#> RECEIVED PUBACK FOR PID 72 (tpc: foo/bar | msg: Round 52) 131
-> /PUBLISH 131
>#>#>#> RECEIVED PUBACK FOR PID 73 (tpc: foo/bar | msg: Round 43) 131
-> /PUBLISH 131
>#>#>#> RECEIVED PUBACK FOR PID 74 (tpc: foo/bar | msg: Round 69) 131
-> /PUBLISH 131
-- MAIN LOOP 132
>#>#>#> RECEIVED PUBACK FOR PID 75 (tpc: foo/bar | msg: Round 47) 132
-> /PUBLISH 132
>#>#>#> RECEIVED PUBACK FOR PID 76 (tpc: foo/bar | msg: Round 67) 132
-> /PUBLISH 132
>#>#>#> RECEIVED PUBACK FOR PID 77 (tpc: foo/bar | msg: Round 36) 132
-> /PUBLISH 132
>#>#>#> RECEIVED PUBACK FOR PID 78 (tpc: foo/bar | msg: Round 31) 132
-> /PUBLISH 132
>#>#>#> RECEIVED PUBACK FOR PID 79 (tpc: foo/bar | msg: Round 48) 132
-> /PUBLISH 132
>#>#>#> RECEIVED PUBACK FOR PID 80 (tpc: foo/bar | msg: Round 65) 132
-> /PUBLISH 132
>#>#>#> RECEIVED PUBACK FOR PID 81 (tpc: foo/bar | msg: Round 56) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 82 (tpc: foo/bar | msg: Round 45) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 83 (tpc: foo/bar | msg: Round 33) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 84 (tpc: foo/bar | msg: Round 35) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 85 (tpc: foo/bar | msg: Round 70) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 86 (tpc: foo/bar | msg: Round 49) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 87 (tpc: foo/bar | msg: Round 46) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 88 (tpc: foo/bar | msg: Round 51) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 89 (tpc: foo/bar | msg: Round 38) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 90 (tpc: foo/bar | msg: Round 68) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 91 (tpc: foo/bar | msg: Round 60) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 92 (tpc: foo/bar | msg: Round 58) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 93 (tpc: foo/bar | msg: Round 63) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 94 (tpc: foo/bar | msg: Round 42) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 95 (tpc: foo/bar | msg: Round 54) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 96 (tpc: foo/bar | msg: Round 50) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 97 (tpc: foo/bar | msg: Round 66) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 98 (tpc: foo/bar | msg: Round 55) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 99 (tpc: foo/bar | msg: Round 57) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 100 (tpc: foo/bar | msg: Round 61) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 101 (tpc: foo/bar | msg: Round 59) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 103 (tpc: foo/bar | msg: Round 39) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 104 (tpc: foo/bar | msg: Round 64) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 102 (tpc: foo/bar | msg: Round 32) 133
-> /PUBLISH 133
>#>#>#> RECEIVED PUBACK FOR PID 105 (tpc: foo/bar | msg: Round 62) 134
-> /PUBLISH 134
>#>#>#> RECEIVED PUBACK FOR PID 106 (tpc: foo/bar | msg: Round 34) 134
-> /PUBLISH 134
>#>#>#> RECEIVED PUBACK FOR PID 107 (tpc: foo/bar | msg: Round 41) 134
-> /PUBLISH 134
>#>#>#> RECEIVED PUBACK FOR PID 108 (tpc: foo/bar | msg: Round 37) 134
-> /PUBLISH 134
>#>#>#> RECEIVED PUBACK FOR PID 109 (tpc: foo/bar | msg: Round 44) 134
-> /PUBLISH 134
-- MAIN LOOP 142
+++ TMR INT TRIGGERED 145
|| Starting asyncfun() 145
|| Scheduling 1st set of 5 messages.. 145
-> PUBLISH 145
-> PUBLISH 145
-> PUBLISH 145
-> PUBLISH 145
-> PUBLISH 145
>#>#>#> RECEIVED PUBACK FOR PID 110 (tpc: foo/bar | msg: Round 71) 146
-> /PUBLISH 146
>#>#>#> RECEIVED PUBACK FOR PID 111 (tpc: foo/bar | msg: Round 72) 146
-> /PUBLISH 146
>#>#>#> RECEIVED PUBACK FOR PID 112 (tpc: foo/bar | msg: Round 73) 146
-> /PUBLISH 146
>#>#>#> RECEIVED PUBACK FOR PID 113 (tpc: foo/bar | msg: Round 74) 146
-> /PUBLISH 146
>#>#>#> RECEIVED PUBACK FOR PID 114 (tpc: foo/bar | msg: Round 75) 146
-> /PUBLISH 146
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 149
|| Scheduling 2nd set of 5 messages.. 149
-> PUBLISH 149
-> PUBLISH 149
-> PUBLISH 149
-> PUBLISH 149
-> PUBLISH 149
>#>#>#> RECEIVED PUBACK FOR PID 115 (tpc: foo/bar | msg: Round 76) 151
-> /PUBLISH 151
>#>#>#> RECEIVED PUBACK FOR PID 116 (tpc: foo/bar | msg: Round 77) 151
-> /PUBLISH 151
>#>#>#> RECEIVED PUBACK FOR PID 117 (tpc: foo/bar | msg: Round 78) 151
-> /PUBLISH 151
>#>#>#> RECEIVED PUBACK FOR PID 118 (tpc: foo/bar | msg: Round 79) 151
-> /PUBLISH 151
>#>#>#> RECEIVED PUBACK FOR PID 119 (tpc: foo/bar | msg: Round 80) 151
-> /PUBLISH 151
-- MAIN LOOP 152
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 153
|| Finished asyncfun() 153
+++ TMR INT TRIGGERED 160
|| Starting asyncfun() 160
|| Scheduling 1st set of 5 messages.. 160
-> PUBLISH 160
-> PUBLISH 160
-> PUBLISH 160
-> PUBLISH 160
-> PUBLISH 160
>#>#>#> RECEIVED PUBACK FOR PID 120 (tpc: foo/bar | msg: Round 81) 161
-> /PUBLISH 161
>#>#>#> RECEIVED PUBACK FOR PID 121 (tpc: foo/bar | msg: Round 82) 161
-> /PUBLISH 161
-- MAIN LOOP 162
>#>#>#> RECEIVED PUBACK FOR PID 122 (tpc: foo/bar | msg: Round 83) 162
-> /PUBLISH 162
>#>#>#> RECEIVED PUBACK FOR PID 123 (tpc: foo/bar | msg: Round 84) 162
-> /PUBLISH 162
>#>#>#> RECEIVED PUBACK FOR PID 124 (tpc: foo/bar | msg: Round 85) 162
-> /PUBLISH 162
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 164
|| Scheduling 2nd set of 5 messages.. 164
-> PUBLISH 164
-> PUBLISH 164
-> PUBLISH 164
-> PUBLISH 164
-> PUBLISH 164
>#>#>#> RECEIVED PUBACK FOR PID 125 (tpc: foo/bar | msg: Round 86) 165
-> /PUBLISH 165
>#>#>#> RECEIVED PUBACK FOR PID 126 (tpc: foo/bar | msg: Round 87) 165
-> /PUBLISH 165
>#>#>#> RECEIVED PUBACK FOR PID 127 (tpc: foo/bar | msg: Round 88) 165
-> /PUBLISH 165
>#>#>#> RECEIVED PUBACK FOR PID 128 (tpc: foo/bar | msg: Round 89) 166
-> /PUBLISH 166
>#>#>#> RECEIVED PUBACK FOR PID 129 (tpc: foo/bar | msg: Round 90) 166
-> /PUBLISH 166
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 168
|| Finished asyncfun() 168
-- MAIN LOOP 172
+++ TMR INT TRIGGERED 175
|| Starting asyncfun() 175
|| Scheduling 1st set of 5 messages.. 175
-> PUBLISH 175
-> PUBLISH 175
-> PUBLISH 175
-> PUBLISH 175
-> PUBLISH 175
>#>#>#> RECEIVED PUBACK FOR PID 130 (tpc: foo/bar | msg: Round 91) 176
-> /PUBLISH 176
>#>#>#> RECEIVED PUBACK FOR PID 131 (tpc: foo/bar | msg: Round 92) 176
-> /PUBLISH 176
>#>#>#> RECEIVED PUBACK FOR PID 132 (tpc: foo/bar | msg: Round 93) 176
-> /PUBLISH 176
>#>#>#> RECEIVED PUBACK FOR PID 133 (tpc: foo/bar | msg: Round 94) 176
-> /PUBLISH 176
>#>#>#> RECEIVED PUBACK FOR PID 134 (tpc: foo/bar | msg: Round 95) 176
-> /PUBLISH 176
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 179
|| Scheduling 2nd set of 5 messages.. 179
-> PUBLISH 179
-> PUBLISH 179
-> PUBLISH 179
-> PUBLISH 179
-> PUBLISH 179
>#>#>#> RECEIVED PUBACK FOR PID 135 (tpc: foo/bar | msg: Round 96) 181
-> /PUBLISH 181
>#>#>#> RECEIVED PUBACK FOR PID 136 (tpc: foo/bar | msg: Round 97) 181
-> /PUBLISH 181
>#>#>#> RECEIVED PUBACK FOR PID 137 (tpc: foo/bar | msg: Round 98) 181
-> /PUBLISH 181
>#>#>#> RECEIVED PUBACK FOR PID 138 (tpc: foo/bar | msg: Round 99) 181
-> /PUBLISH 181
>#>#>#> RECEIVED PUBACK FOR PID 139 (tpc: foo/bar | msg: Round 100) 181
-> /PUBLISH 181
-- MAIN LOOP 182
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 183
|| Finished asyncfun() 183
+++ TMR INT TRIGGERED 190
|| Starting asyncfun() 190
|| Scheduling 1st set of 5 messages.. 190
-> PUBLISH 190
-> PUBLISH 190
-> PUBLISH 190
-> PUBLISH 190
-> PUBLISH 190
>#>#>#> RECEIVED PUBACK FOR PID 140 (tpc: foo/bar | msg: Round 101) 191
-> /PUBLISH 191
>#>#>#> RECEIVED PUBACK FOR PID 141 (tpc: foo/bar | msg: Round 102) 191
-> /PUBLISH 191
>#>#>#> RECEIVED PUBACK FOR PID 142 (tpc: foo/bar | msg: Round 103) 191
-> /PUBLISH 191
>#>#>#> RECEIVED PUBACK FOR PID 143 (tpc: foo/bar | msg: Round 104) 191
-> /PUBLISH 191
>#>#>#> RECEIVED PUBACK FOR PID 144 (tpc: foo/bar | msg: Round 105) 191
-> /PUBLISH 191
-- MAIN LOOP 192
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 194
|| Scheduling 2nd set of 5 messages.. 194
-> PUBLISH 194
-> PUBLISH 194
-> PUBLISH 194
-> PUBLISH 194
-> PUBLISH 194
>#>#>#> RECEIVED PUBACK FOR PID 145 (tpc: foo/bar | msg: Round 106) 196
-> /PUBLISH 196
>#>#>#> RECEIVED PUBACK FOR PID 146 (tpc: foo/bar | msg: Round 107) 196
-> /PUBLISH 196
>#>#>#> RECEIVED PUBACK FOR PID 147 (tpc: foo/bar | msg: Round 108) 196
-> /PUBLISH 196
>#>#>#> RECEIVED PUBACK FOR PID 148 (tpc: foo/bar | msg: Round 109) 196
-> /PUBLISH 196
>#>#>#> RECEIVED PUBACK FOR PID 149 (tpc: foo/bar | msg: Round 110) 196
-> /PUBLISH 196
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 198
|| Finished asyncfun() 198
-- MAIN LOOP 202
+++ TMR INT TRIGGERED 205
|| Starting asyncfun() 205
|| Scheduling 1st set of 5 messages.. 205
-> PUBLISH 205
-> PUBLISH 205
-> PUBLISH 205
-> PUBLISH 205
-> PUBLISH 205
>#>#>#> RECEIVED PUBACK FOR PID 150 (tpc: foo/bar | msg: Round 111) 206
-> /PUBLISH 206
>#>#>#> RECEIVED PUBACK FOR PID 151 (tpc: foo/bar | msg: Round 112) 206
-> /PUBLISH 206
>#>#>#> RECEIVED PUBACK FOR PID 152 (tpc: foo/bar | msg: Round 113) 206
-> /PUBLISH 206
>#>#>#> RECEIVED PUBACK FOR PID 153 (tpc: foo/bar | msg: Round 114) 206
-> /PUBLISH 206
>#>#>#> RECEIVED PUBACK FOR PID 154 (tpc: foo/bar | msg: Round 115) 206
-> /PUBLISH 206
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 209
|| Scheduling 2nd set of 5 messages.. 209
-> PUBLISH 209
-> PUBLISH 209
-> PUBLISH 209
-> PUBLISH 209
-> PUBLISH 209
>#>#>#> RECEIVED PUBACK FOR PID 155 (tpc: foo/bar | msg: Round 116) 211
-> /PUBLISH 211
>#>#>#> RECEIVED PUBACK FOR PID 156 (tpc: foo/bar | msg: Round 117) 211
-> /PUBLISH 211
>#>#>#> RECEIVED PUBACK FOR PID 157 (tpc: foo/bar | msg: Round 118) 211
-> /PUBLISH 211
>#>#>#> RECEIVED PUBACK FOR PID 158 (tpc: foo/bar | msg: Round 119) 211
-> /PUBLISH 211
>#>#>#> RECEIVED PUBACK FOR PID 159 (tpc: foo/bar | msg: Round 120) 211
-> /PUBLISH 211
-- MAIN LOOP 212
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 213
|| Finished asyncfun() 213
+++ TMR INT TRIGGERED 220
|| Starting asyncfun() 220
|| Scheduling 1st set of 5 messages.. 220
-> PUBLISH 220
-> PUBLISH 220
-> PUBLISH 220
-> PUBLISH 220
-> PUBLISH 220
-- MAIN LOOP 222
|| Expected PUBACKs for 1st set of 5 messages to be handled by now 224
|| Scheduling 2nd set of 5 messages.. 224
-> PUBLISH 224
-> PUBLISH 224
-> PUBLISH 224
-> PUBLISH 224
-> PUBLISH 224
|| Expected PUBACKs for 2nd set of 5 messages to be handled by now 228
|| Finished asyncfun() 228
assertion "mp_obj_is_small_int(args[2])" failed: file "../../extmod/moduasyncio.c", line 103, function: task_queue_push_sorted
abort() was called at PC 0x4016b75f on core 0

ELF file SHA256: 0000000000000000

Backtrace: 0x4009ae74:0x3ffce180 0x4009b269:0x3ffce1a0 0x4016b75f:0x3ffce1c0 0x400ee1c6:0x3ffce1f0 0x400e388a:0x3ffce210 0x400df7ed:0x3ffce240 0x400df915:0x3ffce260 0x400ed141:0x3ffce280 0x400e3944:0x3ffce320 0x400df7ed:0x3ffce390 0x400df915:0x3ffce3b0 0x400dfa8b:0x3ffce3d0 0x400ed319:0x3ffce410 0x400e3c67:0x3ffce4b0 0x400e3cc9:0x3ffce4d0 0x400e3d38:0x3ffce500 0x400e37f5:0x3ffce520 0x400df7ed:0x3ffce540 0x400df915:0x3ffce560 0x400ed141:0x3ffce580 0x400e3944:0x3ffce620 0x400df7ed:0x3ffce650 0x400ed0b9:0x3ffce670 0x400e3944:0x3ffce710 0x400df7ed:0x3ffce780 0x400df915:0x3ffce7a0 0x400ed141:0x3ffce7c0 0x400e3944:0x3ffce860 0x400df7ed:0x3ffce8c0 0x400df816:0x3ffce8e0 0x400df88b:0x3ffce900 0x400ea521:0x3ffce990 0x400ea857:0x3ffce9c0 0x400df992:0x3ffceaa0 0x400ed3a5:0x3ffceae0 0x400e3944:0x3ffceb80 0x400df7ed:0x3ffcebe0 0x400df816:0x3ffcec00 0x40104c52:0x3ffcec20 0x40104f74:0x3ffcecb0 0x400f60dc:0x3ffcecf0 0x4009b831:0x3ffced20

Rebooting...
MicroPython v1.14 on 2021-02-02; ESP32 module with ESP32
Type "help()" for more information.

async `subs_cb`?

Presumably it's very common to publish an MQTT message in response to receiving one (after changing some internal state, for example). Since publish is async, I was surprised that subs_cb is not async, so that you cannot await anything that publishes.

I presume a workaround is something like:

def cb(client):
    uasyncio.create_task(some_async_function())

but then you can't wait, catch exceptions, etc.

Error breaking outage resilience on ESP32: "wifi:AP has neither DSSS parameter nor HT Information, drop it"

I encountered this error while testing my code for resilience against internet/wifi outage...
It starts with a ECONNABORTED and retries, failing with ENOTCONN just like expected.

At a certain point though, it throw the wifi:AP error once or twice and then freeze.
Since I have other coroutines running and they are under watchdogs, the fact that it froze means it simply exited the _keep_connected coroutine.

My guess is it's the wifi_connect call inside it that now throws a non-OSError in MicroPython 1.17

Error in reconnect. [Errno 128] ENOTCONN
Connecting to broker.
Error in reconnect. [Errno 128] ENOTCONN
Connecting to broker.
Error in reconnect. [Errno 128] ENOTCONN
Connecting to broker.
Error in reconnect. [Errno 128] ENOTCONN
E (319425) wifi:AP has neither DSSS parameter nor HT Information, drop it
E (319615) wifi:AP has neither DSSS parameter nor HT Information, drop it
Traceback (most recent call last):
...
...
KeyboardInterrupt: 
MicroPython v1.17 on 2021-09-02; ESP32 module (spiram) with ESP32
Type "help()" for more information.

Every time I kill internet connectivy, it errors out once.
When I kill wifi, it errors twice before freezing.

Edit:
It's not exiting the wifi_connect(), its stuck here:

else:
  while s.status() == network.STAT_CONNECTING:  # Break out on fail or success. Check once per sec.
      await asyncio.sleep(1)

Return to main application when errors are found

Hi.

This library works like a charm.
My only comment is in the function:

async def _keep_connected(self):

When there is an error, for instance, If the broker is not available, the client.publish does not ever return.

This is bad since the client may need to do other stuff, such as store the payload to the disk and retry later.

Can we add one more argument and tell the library to generate an exception so the client can take the actions accordingly?

Or add a timeout in:

    # Await broker connection.
    async def _connection(self):
        while not self._isconnected:
            await asyncio.sleep(1)

Thanks

esp8266 keep restarting

Hi
I cross compiled mqtt_as.py with mpy_cross and use it in my project on esp8266, this is my code and my problem is after i deploy main.py to board and restart it, keep restarting untile i kill it !

async def conn_han(client):
    print(">>>>>>>>> inside conn_han")
    await client.subscribe('foo_topic', 1)

async def mqtt_connector(client):
    print(">> connecting to MQTT Broker ...")
    await client.connect()
    print(">> connected MQTT Broker .")

if __name__ == "__main__":
...
config = {
        'client_id': mqtt_client_id,
        'server': mqtt_server_host,
        'port': mqtt_server_port,
        'user': mqtt_server_username,
        'password': mqtt_server_password,
        'keepalive': 60,
        'ping_interval': 0,
        'ssl': True,
        'ssl_params': {},
        'response_time': 10,
        'clean_init': True,
        'clean': True,
        'max_repubs': 4,
        'will': None,
        'subs_cb': lambda *_: None,
        'wifi_coro': "",
        'connect_coro': conn_han,
        'wifi_pw': None,
    }
    mqtt_client = MQTTClient(config)
    try:
        loop.run_until_complete(mqtt_connector(client=mqtt_client))
    finally:
        mqtt_client.close()  # Prevent LmacRxBlk:1 errors

>> connecting to MQTT Broker ...

 ets Jan  8 2013,rst cause:2, boot mode:(3,0)

load 0x40100000, len 31024, room 16 
tail 0
chksum 0xb4
load 0x3ffe8000, len 1104, room 8 
tail 8
chksum 0x74
load 0x3ffe8450, len 824, room 0 
tail 8
chksum 0x45
csum 0x45
���2�o�{��n|�
             $
              l$l b���
                      �|s��$�N��o�l`��r�p��l�

                                             $`��r�p��l�
                                                        $`��r�l���
                                                                  l$`rl��o���B
                                                                              ��B|
                                                                                  $b���#|����ll#��o�n�WebREPL daemon started 

I'm connected to wifi through boot.py and there is no need to do it in mqtt_as because of this i not fill wifi setting in config .

Correct Error Handling for reconnect (better than range.py?)

Adopting the error handling in the example code from range.py seems to obviate the reconnection behaviours of micropython-mqtt

Is there an alternative reference example for invoking micropython-mqtt to include error-handling for long-lived applications where transient disconnections are possible? What is the proper invocation to combine a handful of scheduled async functions in such a way that the (implicit) async functions of micropython-mqtt have their errors handled properly, and the application doesn't terminate?

I created a routine based on range.py, but after a very long, and visibly successful test run (e.g. between 20hours and 2 days), it seems to hit unrecoverable connection errors which look like pages and pages of...

LmacRxBlk:1
LmacRxBlk:1
LmacRxBlk:1
LmacRxBlk:1

...followed by cycles of...

state: 5 -> 2 (2c0)
rm 0
LmacRxBlk:1
reconnect
state: 2 -> 0 (0)
scandone
state: 0 -> 2 (b0)
state: 2 -> 3 (0)
state: 3 -> 5 (10)
add 0
aid 1
cnt 
LmacRxBlk:1
LmacRxBlk:1
LmacRxBlk:1
LmacRxBlk:1
state: 5 -> 2 (2c0)

And the application makes no apparent attempt to resume.

Previously though failures happened, the logic of micropython-mqtt seemed to negotiate a reconnect. The change as far as I can see; I added back the finally clause (as per the original example code at range.py ) to trigger client.close() for KeyboardInterrupt (see this commit).

This avoids LmacRxBlk errors when testing interactively (handles keyboard interrupts more gracefully without the repeated errors spewing to the console), but I suspect the finally and client.close() is also triggered when there is a socket error, which then prevents the system from being able to auto-recover from a disconnection.

I am not sure what triggers the disconnection and it seems probabilistic. There may be some configuration option which would improve things. I am not certain whether access-point mode is fully disabled by the invocations I am making, for example. However, I certainly wouldn't expect the ESP8266 to be 'out of range' given the access point is just a couple of metres away.

The 2metreX2metre display, driven by a single ESP8266 (Wemos D1 Mini or equivalent clone), can be seen running in real-time at https://photos.app.goo.gl/14ryl7bCXeYJbNdy2 which gives you an idea of the data rates involved - just the information for 16 segments of 3 'RGB' bytes re-published lazily (only when segments change), and then forwarded by serial from Micropython to an Arduino which controls the lights.

The issue of lost connection can always be resolved by a hard reboot and the display can be back very quickly on reset, then runs for another several hours. However, I have to eliminate the need for a manual reset, and would like the error recovery and any reset to be handled in software instead, and hopefully therefore a bit faster. This is intended to be part of a display of 20 characters, so if they all have a MTBF of 20 hours, that means a glitch every hour. If we encounter any environmental factors which might trigger wifi/network errors more frequently than once a day, I definitely need better error-handling. The shorter and rarer I can get failures to be, the better.

QUESTIONS

I seem to recall there is a fallback option on ESP8266 of wiring an output pin to its own reset, and triggering a hard reset that way. In your experience, is a 'hardware self-reset' the best way to conceive of a long-lived networked device which might not occasionally enter an unrecoverable loop of errors from some OS-level network issue? Seems heavy-handed for just recovering a transient wifi or socket error.

Is there a way to avoid these networking errors in the first place by configuring the application differently (e.g. turning off some Wifi feature)?

Finally is there a better way to instruct the handling of these errors than the finally clause in range.py. Ideally error handling which is designed to keep the application running rather than terminate it?

Application background information

My code was derived from the range.py example, but adding a bit of translation between byte frames dispatched over MQTT and serial frames sent over to 5V Arduino-compatible circuit. The Arduino acts as a 'co-processor' to service large arrays of WS2811 addressable LEDs, but could also potentially take over the role of a reset watchdog if really necessary (e.g. power cycle the ESP8266 if it doesn't get anything over serial for a while).

My current Micropython code is at uartcharactertest.py and the Arduino code is at uartcharacter.ino (although the ESP8266 receive pin isn't even wired, so the Arduino code is strictly 'downstream' and can't have any effect on the micropython board).

The controller (MQTT publisher) and MQTT broker are running on a Pi 3. They remain functional throughout and have run for many days at a time without visible fault (as demonstrated by other desktop-based subscribers to the display information). The wifi network is served by a GL iNet AR150 running OpenWRT which has also never failed so far. Repowering the micropython board is enough to bring back full function in all cases I have encountered, meaning the error exists somewhere in the ESP8266 or Micropython networking stack, or in my error handling (based on range.py).

The raw log of a failure is visible at https://raw.githubusercontent.com/cefn/retrotextual/master/gist/cockle-longrun-networking-error.log - search for LmacRxBlk to find the start of the failure.

How to subscribe in subs_cb?

Hi all,

I try to setup an ESP32 which get its config from a mqtt topic.
So I subscribe the topic in the connect_coro.
When the message comes I want to subscribe other topics according to this message.
But how can I subscribe something in the subs_cb? It is no coroutine and when I try to await mqtt.subscribe() inside the subs_cb it isn't called at all.

What would be the right way to do that?

Question : Callback to function outside of own class

I'm not sure if this question should be asked here but I have the following situation.

I have 2 async tasks running from the main loop :

• UartConnection (in file .UartHelper)
• MqttClass (in file .MqttComp)

these are both living in their own file. So the main loop looks like this:

from .UartHelper import UartConnection
from .MqttComp import MqttClass, config
import uasyncio as asyncio


class Main():
    def __init__(self):
        self.uart_interface = UartConnection()
        self.mqtt = MqttClass()

    async def loop(self):
        await self.mqtt.start_connection()
        # get UART data
        while True:
            uart_connection_task = asyncio.create_task(self.uart_interface.check_uart())
            current_uart_state = await uart_connection_task
            asyncio.create_task(self.mqtt.publish_states(states=current_uart_state))

            await asyncio.sleep(1)

i am able to use self.config['subs_cb'] in self.mqtt for setting a callback function but this function seems to only work if it is inside the MqttClass. What I basically want is the following :

Mqtt launches the callback function -> MqttClass asks main class to run function from UartConnection class (Send Uart command)

how would I achieve this? Since I also want to add a web page control for this I prefer to not use the MQTT class as main class.

SSL issues

I am trying to move over my project onto this library, but am having some issues with SSL. I see that you didn't have much luck, but these settings work with MQTT.robust (albeit slowly). I've tried searching for ssl_handshake_status: -4 to no avail.

The code:

from mqtt_as import MQTTClient
from mqtt_as import config
import uasyncio as asyncio

CERT_FILE = "/flash/8266-01.cert.der"
KEY_FILE = "/flash/8266-01.key.der"
#if you change the ClientId make sure update AWS policy
MQTT_CLIENT_ID = "basicPubSub"
MQTT_PORT = 8883
#if you change the topic make sure update AWS policy
MQTT_TOPIC = "sdk/test/Python"
MQTT_HOST = "amazon endpoint"

try:
    with open(KEY_FILE, "r") as f: 
        key = f.read()
#         print("Got Key")
        
    with open(CERT_FILE, "r") as f: 
        cert = f.read()
#         print("Got Cert")	
except:
    print('oh fuck')
        
def callback(topic, msg):
    print((topic, msg))

async def conn_han(client):
    await client.subscribe('foo_topic', 1)

async def main(client):
    await client.connect()
    n = 0
    while True:
        await asyncio.sleep(5)
        print('publish', n)
        # If WiFi is down the following will pause for the duration.
        await client.publish('result', '{}'.format(n), qos = 1)
        n += 1

config['subs_cb'] = callback
config['connect_coro'] = conn_han
config['server'] = MQTT_HOST
config['port'] = MQTT_PORT
config['ssl'] = True
config['ssl_params'] = {"cert":cert, "key":key, "server_side":False}

MQTTClient.DEBUG = True  # Optional: print diagnostic messages
client = MQTTClient(config)
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main(client))
finally:
    client.close()  # Prevent LmacRxBlk:1 errors

Here's the error:

Connecting to broker.
ssl_handshake_status: -4
Traceback (most recent call last):
  File "<stdin>", line 57, in <module>
  File "<stdin>", line 55, in <module>
  File "uasyncio/core.py", line 187, in run_until_complete
  File "uasyncio/core.py", line 161, in run_forever
  File "uasyncio/core.py", line 116, in run_forever
  File "uasyncio/core.py", line 184, in _run_and_stop
  File "<stdin>", line 35, in main
  File "mqtt_as.py", line 514, in connect
  File "mqtt_as.py", line 246, in _connect

Unicode letter in MQTT topic or message e.g. "°" lets crash publish()

This is reproducible output when having UTF letter in topic or msg:
WiFi not connected
Checking WiFi integrity.
Got reliable connection with IP 192.168.1.4
Connecting to broker.
Connected to broker.
Wifi is up
publish topic + msg ['test/powersensor/temperature/°$unit', '°C']
Wifi is down
Connection error
WiFi not connected
Checking WiFi integrity.

In source code I could see that the length is derived from msg before sending. May be the issue.

Working with a wired LAN

Will the library work with a wired network, such as a ESP32 + ethernet LAN8720A?

Will not sending update

I have will enabled in my config.py as such:
config['will'] = ('myfeed', 'Offline', True, 1)

When it goes offline this feed never is seen on the server.

disconnect() doesn't work, triggers reconnect

The disconnect method doesn't work because it just closes the socket but keep_connected is still running and triggers a reconnect. _has_connected would need to be reset and _keep_connected canceled.
I might create a PR the next days if you like.

Ability to set multiple wifi ssids and passwords in the configuration

If ssid is of type list, then it could scan the network for the strongest signal in this list before connecting to it.

This would mean I'd have portable devices that could be programmed to connect with lots of wifi networks in different locations. This feature would make life as convenient as having my laptop automatically connecting to a different network when I'm in my office than at home without having to manually select it.

(It needs to select for the strongest signal, rather than the first match it finds, otherwise it's very difficult to control. When I did this with two devices that both knew the wifis of two phones I had to carry one phone and device to the other end of a field so I could get the other two devices to pair up properly.)

keepalive race condition

Another nit ;-)

Unless the user explicitly sets config['ping_interval'] the timeout where the _keep_alive starts a reconnection process is the same as the keep alive value set at the MQTT protocol level. This produces a bit of a race condition as to whom times out first.

Unless I'm missing something, the two effects of the keep alive mechanism at the MQTT level are for the broker to send the last will message when it discovers that the keep live timed out, and for the broker to close the connection when 1.5x the keep-alive times out. Nothing else.

So when the _keep_alive coro discovers a time-out there's a bit of a race condition between the client reconnecting and the broker discovering the time-out and triggering the last will message. If no last will message is set, the MQTT keep alive mechanism is really pretty irrelevant given the client-side timeout. IMHO for proper last will operation the coro's time-out should be shorter than the MQTT keep-alive time, which can be set explicitly using the ping_interval config param but that's not exactly obvious to someone that just naively sets the last will message.

config.clean_init vs. config.clean

This is not a bug report, just a little heads-up ;-)

I'm not sure I understand correctly what the intent of these two flags is, but I don't believe they function as intended. My understanding is that having two flags is intended to allow a mode (clean_init=true, clean=false) where the initial connection creates a clean state but then subsequent automatic reconnections continue with the previous state such that unacked packets queued at the broker get retransmitted. This seems to be what unclean.py demonstrates.

However, section 3.1.2.4 of the MQTT 3.1.1 spec states:

If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one. This Session lasts as long as the Network Connection. State data associated with this Session MUST NOT be reused in any subsequent Session.

Note the last sentence: when the initial clean connection ends, the state, including queued packets and subscriptions, must be deleted by the broker.

The unclean.py example sort-a works because it does not notice that unacked packets from the first connection are lost and because it resubscribes on each subsequent connection even though that is not necessary if the connection state were always retained.

Can not access to ESP8266 if Wifi out of range

Hi,

I've noticed that I could not upload to my board (ESP8266) if my wifi network was out of range.

ampy.pyboard
.PyboardError: could not exec command

If I come closer to the setup wifi, all works fine again. In some cases my boards are programmed on different wifi networks and I would have flash them do to recover access.

Best,

Sébastien