aiomqtt raise: Operation timed out #254

zarch · 2023-10-25T10:22:27Z

zarch
Oct 25, 2023

Dear all,

I really like the interface that you have developed.
However, I'm facing an issue that I'm not able to fix.

If I use the paho Client directly is working, but it is raising an exception when using aiomqtt.

import asyncio
import os
import random
from datetime import datetime

import paho.mqtt.client as mqtt
from aiomqtt import Client, MqttError
from loguru import logger
from pytz import timezone

tz = timezone("Europe/Rome")

MQTT_HOST = os.environ["MQTT_HOST"]
MQTT_PORT = int(os.environ["MQTT_PORT"])
MQTT_VHOST = os.environ["MQTT_VHOST"]
MQTT_USER = os.environ["MQTT_USER"]
MQTT_PASSWORD = os.environ["MQTT_PASSWORD"]
MQTT_TOPIC = os.environ["MQTT_TOPIC"]
USERNAME = f"{MQTT_VHOST}:{MQTT_USER}"


def gen_payload(sender: str) -> str:
    """Generate random payload."""
    time = datetime.now(tz)
    return (
        f'{{"sender":"{sender}", '
        f'"value":{random.random() * 100:.2f}, '
        f'"time":"{time.isoformat()}"}}'
    )


def send_via_paho() -> None:
    """Send value via paho."""

    def on_connect(client, userdata, flags, rc) -> None:  # noqa: ANN001, ARG001
        logger.info("Connected with result code " + str(rc))
        client.subscribe("$SYS/#")

    def on_message(client, userdata, msg) -> None:        # noqa: ANN001, ARG001
        logger.info(msg.topic + " " + str(msg.payload))

    def on_publish(mosq, obj, mid) -> None:               # noqa: ANN001, ARG001
        logger.info("mid: " + str(mid))

    def on_subscribe(mosq, obj, mid, granted_qos) -> None:  # noqa: ANN001, ARG001
        logger.info("Subscribed: " + str(mid) + " " + str(granted_qos))

    def on_log(mosq, obj, level, string) -> None:         # noqa: ANN001, ARG001
        logger.info(string)

    client = mqtt.Client()

    # Assign event callbacks
    client.on_message = on_message
    client.on_connect = on_connect
    client.on_publish = on_publish
    client.on_subscribe = on_subscribe

    topic = f"{MQTT_TOPIC}/sys/cpu/percent"
    payload = gen_payload("paho")

    client.username_pw_set(username=USERNAME, password=MQTT_PASSWORD)
    client.connect(host=MQTT_HOST, port=MQTT_PORT, keepalive=60)
    client.subscribe(topic, 1)

    msginfo = client.publish(topic=topic, payload=payload)
    msginfo.wait_for_publish()
    if msginfo.rc != 0:
        # MQTTClient_connect()
        # https://www.eclipse.org/paho/files/mqttdoc/MQTTClient/html/_m_q_t_t_client_8h.html#aaa8ae61cd65c9dc0846df10122d7bd4e
        err_code = {
            1: "Connection refused: Unacceptable protocol version",
            2: "Connection refused: Identifier rejected",
            3: "Connection refused: Server unavailable",
            4: "Connection refused: Bad user name or password",
            5: "Connection refused: Not authorized",
        }
        emsg = (
            err_code[msginfo.rc]
            if msginfo.rc in err_code
            else "Error code not handled yet..."
        )
        logger.warning(f"returned code: {msginfo.rc}: {emsg}")
    client.disconnect()
    logger.info("DONE!")


async def send_via_aiomqtt() -> None:
    """Send value via aiomqtt."""
    mqtt_client = Client(
        hostname=MQTT_HOST,
        port=MQTT_PORT,
        username=USERNAME,
        password=MQTT_PASSWORD,
        logger=logger,
        timeout=10,
    )

    topic = f"{MQTT_TOPIC}/sys/cpu/percent"
    payload = gen_payload("aiomqtt")
    async with mqtt_client as client:
        logger.info(f"Subscribing to: {topic}")
        await client.subscribe(topic, qos=2)
        logger.info(f"Publishing to: {topic} -> {payload}")
        await client.publish(topic, payload=payload, qos=2)
    logger.info("DONE!")


if __name__ == "__main__":
    send_via_paho()

    asyncio.run(send_via_aiomqtt())

When executing the code I see:

python mq.py
2023-10-25 12:14:08.122 | INFO     | __main__:on_publish:61 - mid: 2
2023-10-25 12:14:08.122 | INFO     | __main__:send_via_paho:103 - DONE!
2023-10-25 12:14:08.156 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending CONNECT (u%d, p%d, wr%d, wq%d, wf%d, c%d, k%d) client_id=%s
2023-10-25 12:14:08.178 | Level 10 | paho.mqtt.client:_easy_log:2529 - Received CONNACK (%s, %s)
2023-10-25 12:14:08.179 | INFO     | __main__:send_via_aiomqtt:120 - Subscribing to: g7OARv2bNZw/sys/cpu/percent
2023-10-25 12:14:08.179 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending SUBSCRIBE (d%d, m%d) %s
2023-10-25 12:14:08.212 | Level 10 | paho.mqtt.client:_easy_log:2529 - Received SUBACK
2023-10-25 12:14:08.212 | INFO     | __main__:send_via_aiomqtt:122 - Publishing to: g7OARv2bNZw/sys/cpu/percent -> {"sender":"aiomqtt", "value":95.50, "time":"2023-10-25T12:14:08.123741+02:00"}
2023-10-25 12:14:08.213 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending PUBLISH (d%d, q%d, r%d, m%d), '%s', ... (%d bytes)
Traceback (most recent call last):
  File "/data/src/syn/syngw/gw/mq.py", line 123, in send_via_aiomqtt
    await client.publish(topic, payload=payload, qos=2)
  File "/data/src/syn/syngw/.venv/lib/python3.11/site-packages/aiomqtt/client.py", line 117, in decorated
    return await method(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/data/src/syn/syngw/.venv/lib/python3.11/site-packages/aiomqtt/client.py", line 687, in publish
    await self._wait_for(confirmation.wait(), timeout=timeout)
  File "/data/src/syn/syngw/.venv/lib/python3.11/site-packages/aiomqtt/client.py", line 872, in _wait_for
    raise MqttError(msg) from None
aiomqtt.error.MqttError: Operation timed out

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/data/src/syn/syngw/gw/mq.py", line 130, in <module>
    asyncio.run(send_via_aiomqtt())
  File "/usr/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/data/src/syn/syngw/gw/mq.py", line 119, in send_via_aiomqtt
    async with mqtt_client as client:
  File "/data/src/syn/syngw/.venv/lib/python3.11/site-packages/aiomqtt/client.py", line 1107, in __aexit__
    raise exc
aiomqtt.error.MqttCodeError: [code:7] The connection was lost.

Reading the messages with mosquitto I see:

$ mosquitto_sub -h $MQTT_HOST -p $MQTT_PORT -u $MQTT_VHOST:$MQTT_USER -P $MQTT_PASSWORD -t "$MQTT_TOPIC/sys/#" --pretty -v

g7OARv2bNZw/sys/cpu/percent {"sender":"paho", "value":69.59, "time":"2023-10-25T12:14:07.998301+02:00"}

I tried several time but the aiomqtt version is always disconnecting.
What I'm missing?

empicano · 2024-01-23T00:59:58Z

empicano
Jan 23, 2024
Maintainer

Hi there Pietro,

Sorry for the late reply, this discussion slipped through the cracks it seems. I hope it's not too late for some thoughts.

If I understand things right, the same message arrives at the broker with paho but fails to arrive with aiomqtt. The timeout basically tells us that we tried to send the message for 10 seconds, and then gave up.

What caught my eye is the code sends messages with QoS=0 for paho and QoS=2 for aiomqtt. Does the issue persist if you send messages with QoS=0 for aiomqtt as well?

2 replies

zarch Feb 15, 2024
Author

Hi @empicano,

I've somehow miss your replay... :-D
Good catch, I didn't notice that the QoS was different.

Summary

I rewrote the code to take the Qos as parameter from the command line,
The main results are:

--qos=0 both functions are working;
--qos=1 paho is working while aiomqtt is stack after few minutes I've interrupt the process with Ctrl+C;
--qos=2 both functions are NOT working, as before after few minutes I interrupt the script;

Do you have an idea on why changing the QoS change the result of the task? Any other hint? :-)
Anyway thank you for the amazing work that you are doing and to take time to reply to me!

Code

The main change are:

I added a function to get an ArgumentParser instance to define the QoS from commandline;
I tried to be consistent in both library to use always the same lavel of QoS.

The diff are:

0a1,2
> """Testing paho / aiomqtt behaviour."""
> import argparse
22c24,34
< def gen_payload(sender: str) -> str:
---
> def get_parser() -> argparse.ArgumentParser:
>     """Add QoS parameter handler."""
>     parser = argparse.ArgumentParser(
>         prog="mq",
>         description="Testing different paho MQTT libraries",
>     )
>     parser.add_argument("-q", "--qos", type=int, required=True, metavar="QoS")
>     return parser
>
>
> def gen_payload(sender: str, qos: int) -> str:
26a39
>         f'"qos":{qos}, '
32c45
< def send_via_paho() -> None:
---
> def send_via_paho(qos: int) -> None:
60c73
<     payload = gen_payload("paho")
---
>     payload = gen_payload("paho", qos)
64c77
<     client.subscribe(topic, 1)
---
>     client.subscribe(topic, qos=qos)
66c79
<     msginfo = client.publish(topic=topic, payload=payload)
---
>     msginfo = client.publish(topic=topic, payload=payload, qos=qos)
88c101
< async def send_via_aiomqtt() -> None:
---
> async def send_via_aiomqtt(qos: int) -> None:
96c109
<         timeout=10,
---
>         timeout=3,
100c113
<     payload = gen_payload("aiomqtt")
---
>     payload = gen_payload("aiomqtt", qos)
103c116
<         await client.subscribe(topic, qos=2)
---
>         await client.subscribe(topic, qos=qos)
105c118
<         await client.publish(topic, payload=payload, qos=2)
---
>         await client.publish(topic, payload=payload, qos=qos)
110c123,125
<     send_via_paho()
---
>     parser = get_parser()
>     args = parser.parse_args()
>     send_via_paho(args.qos)
112c127
<     asyncio.run(send_via_aiomqtt())
---
>     asyncio.run(send_via_aiomqtt(args.qos))

The resulting code is now:

"""Testing paho / aiomqtt behaviour."""
import argparse
import asyncio
import os
import random
from datetime import datetime

import paho.mqtt.client as mqtt
from aiomqtt import Client, MqttError
from loguru import logger
from pytz import timezone

tz = timezone("Europe/Rome")

MQTT_HOST = os.environ["MQTT_HOST"]
MQTT_PORT = int(os.environ["MQTT_PORT"])
MQTT_VHOST = os.environ["MQTT_VHOST"]
MQTT_USER = os.environ["MQTT_USER"]
MQTT_PASSWORD = os.environ["MQTT_PASSWORD"]
MQTT_TOPIC = os.environ["MQTT_TOPIC"]
USERNAME = f"{MQTT_VHOST}:{MQTT_USER}"


def get_parser() -> argparse.ArgumentParser:
    """Add QoS parameter handler."""
    parser = argparse.ArgumentParser(
        prog="mq",
        description="Testing different paho MQTT libraries",
    )
    parser.add_argument("-q", "--qos", type=int, required=True, metavar="QoS")
    return parser


def gen_payload(sender: str, qos: int) -> str:
    """Generate random payload."""
    time = datetime.now(tz)
    return (
        f'{{"sender":"{sender}", '
        f'"qos":{qos}, '
        f'"value":{random.random() * 100:.2f}, '
        f'"time":"{time.isoformat()}"}}'
    )


def send_via_paho(qos) -> None:
    """Send value via paho."""

    def on_connect(
        client, userdata, flags, rc
    ) -> None:  # noqa: ANN001, ARG001
        logger.info("Connected with result code " + str(rc))
        client.subscribe("$SYS/#")

    def on_message(client, userdata, msg) -> None:  # noqa: ANN001, ARG001
        logger.info(msg.topic + " " + str(msg.payload))

    def on_publish(mosq, obj, mid) -> None:  # noqa: ANN001, ARG001
        logger.info("mid: " + str(mid))

    def on_subscribe(
        mosq, obj, mid, granted_qos
    ) -> None:  # noqa: ANN001, ARG001
        logger.info("Subscribed: " + str(mid) + " " + str(granted_qos))

    def on_log(mosq, obj, level, string) -> None:  # noqa: ANN001, ARG001
        logger.info(string)

    client = mqtt.Client()

    # Assign event callbacks
    client.on_message = on_message
    client.on_connect = on_connect
    client.on_publish = on_publish
    client.on_subscribe = on_subscribe

    topic = f"{MQTT_TOPIC}/sys/cpu/percent"
    payload = gen_payload("paho", qos)

    client.username_pw_set(username=USERNAME, password=MQTT_PASSWORD)
    client.connect(host=MQTT_HOST, port=MQTT_PORT, keepalive=60)
    client.subscribe(topic, qos=qos)

    msginfo = client.publish(topic=topic, payload=payload, qos=qos)
    msginfo.wait_for_publish()
    if msginfo.rc != 0:
        # MQTTClient_connect()
        # https://www.eclipse.org/paho/files/mqttdoc/MQTTClient/html/_m_q_t_t_client_8h.html#aaa8ae61cd65c9dc0846df10122d7bd4e
        err_code = {
            1: "Connection refused: Unacceptable protocol version",
            2: "Connection refused: Identifier rejected",
            3: "Connection refused: Server unavailable",
            4: "Connection refused: Bad user name or password",
            5: "Connection refused: Not authorized",
        }
        emsg = (
            err_code[msginfo.rc]
            if msginfo.rc in err_code
            else "Error code not handled yet..."
        )
        logger.warning(f"returned code: {msginfo.rc}: {emsg}")
    client.disconnect()
    logger.info("DONE!")


async def send_via_aiomqtt(qos) -> None:
    """Send value via aiomqtt."""
    mqtt_client = Client(
        hostname=MQTT_HOST,
        port=MQTT_PORT,
        username=USERNAME,
        password=MQTT_PASSWORD,
        logger=logger,
        timeout=3,
    )

    topic = f"{MQTT_TOPIC}/sys/cpu/percent"
    payload = gen_payload("aiomqtt", qos)
    async with mqtt_client as client:
        logger.info(f"Subscribing to: {topic}")
        await client.subscribe(topic, qos=qos)
        logger.info(f"Publishing to: {topic} -> {payload}")
        await client.publish(topic, payload=payload, qos=qos)
    logger.info("DONE!")


if __name__ == "__main__":
    parser = get_parser()
    args = parser.parse_args()
    send_via_paho(args.qos)
    asyncio.run(send_via_aiomqtt(args.qos))

QoS = 0

With a QoS=0 I have:

 🐍 v3.12.2 (.venv)
❯ python mq.py -q 0
2024-02-15 10:37:55.787 | INFO     | __main__:on_publish:58 - mid: 2
2024-02-15 10:37:55.788 | INFO     | __main__:send_via_paho:102 - DONE!
2024-02-15 10:37:55.878 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending CONNECT (u%d, p%d, wr%d, wq%d
, wf%d, c%d, k%d) client_id=%s
2024-02-15 10:37:56.585 | Level 10 | paho.mqtt.client:_easy_log:2529 - Received CONNACK (%s, %s)
2024-02-15 10:37:56.586 | INFO     | __main__:send_via_aiomqtt:119 - Subscribing to: g7OARv2bNZw/sys/
cpu/percent
2024-02-15 10:37:56.586 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending SUBSCRIBE (d%d, m%d) %s
2024-02-15 10:37:56.663 | Level 10 | paho.mqtt.client:_easy_log:2529 - Received SUBACK
2024-02-15 10:37:56.664 | INFO     | __main__:send_via_aiomqtt:121 - Publishing to: g7OARv2bNZw/sys/c
pu/percent -> {"sender":"aiomqtt", "qos":0, "value":2.42, "time":"2024-02-15T10:37:55.789021+01:00"}
2024-02-15 10:37:56.664 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending PUBLISH (d%d, q%d, r%d, m%d),
 '%s', ... (%d bytes)
2024-02-15 10:37:56.665 | Level 10 | paho.mqtt.client:_easy_log:2529 - Sending DISCONNECT
2024-02-15 10:37:56.665 | INFO     | __main__:send_via_aiomqtt:123 - DONE!

Checking the message with mosquitto I see:

g7OARv2bNZw/sys/cpu/percent {"sender":"paho", "qos":0, "value":82.90, "time":"2024-02-15T10:37:55.77
0154+01:00"}
g7OARv2bNZw/sys/cpu/percent {"sender":"aiomqtt", "qos":0, "value":2.42, "time":"2024-02-15T10:37:55.
789021+01:00"}

Very good!

QoS = 1

With a QoS=1 I have:

 🐍 v3.12.2 (.venv)
❯ python mq.py -q 1

^CTraceback (most recent call last):
  File "/data/src/syn/syngw/mq.py", line 129, in <module>
    send_via_paho(args.qos)
  File "/data/src/syn/syngw/mq.py", line 84, in send_via_paho
    msginfo.wait_for_publish()
  File "/data/src/syn/syngw/.venv/lib/python3.12/site-packages/paho/mqtt/client.py", line 371, in wait_for_p
ublish
    self._condition.wait(timeout_tenth)
  File "/usr/lib/python3.12/threading.py", line 355, in wait
    waiter.acquire()
KeyboardInterrupt

g7OARv2bNZw/sys/cpu/percent {"sender":"paho", "qos":1, "value":94.82, "time":"2024-02-15T10:38:03.32
8874+01:00"}

QoS = 2

With a QoS=2 I have:

 🐍 v3.12.2 (.venv)
❯ python mq.py -q 2
^CTraceback (most recent call last):
  File "/data/src/syn/syngw/mq.py", line 129, in <module>
    send_via_paho(args.qos)
  File "/data/src/syn/syngw/mq.py", line 84, in send_via_paho
    msginfo.wait_for_publish()
  File "/data/src/syn/syngw/.venv/lib/python3.12/site-packages/paho/mqtt/client.py", line 371, in wait_for_publish
    self._condition.wait(timeout_tenth)
  File "/usr/lib/python3.12/threading.py", line 355, in wait
    waiter.acquire()
KeyboardInterrupt

No message reach the MQTT server.

empicano Feb 18, 2024
Maintainer

Hi Pietro,

Thanks for checking back and the detailed description!

I tried your example slightly modified with the "test.mosquitto.org" broker. When I switch the two lines ...

send_via_paho(args.qos)
asyncio.run(send_via_aiomqtt(args.qos))

... to run the aiomqtt code before the paho-mqtt code, the aiomqtt code runs fine for me. Somehow the callbacks in the paho-mqtt code example are never called, meaning the message is sent, but we never receive an answer from the broker. That's why it seems to get stuck. Maybe you can dig around why that is?

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

aiomqtt raise: Operation timed out #254

{{title}}

Replies: 1 comment 2 replies

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

Select a reply

aiomqtt raise: Operation timed out #254

zarch Oct 25, 2023

Replies: 1 comment · 2 replies

empicano Jan 23, 2024 Maintainer

zarch Feb 15, 2024 Author

Summary

Code

QoS = 0

QoS = 1

QoS = 2

empicano Feb 18, 2024 Maintainer

zarch
Oct 25, 2023

Replies: 1 comment 2 replies

empicano
Jan 23, 2024
Maintainer

zarch Feb 15, 2024
Author

empicano Feb 18, 2024
Maintainer