Client does not reconnect

[naquad]

Hi,

I'm using version 0.5.6 and got the following code:

#!/usr/bin/env python3

import asyncio
from gmqtt import Client as MQTTClient
from gmqtt.mqtt.constants import UNLIMITED_RECONNECTS
import logging

logging.basicConfig(level=logging.DEBUG)

async def main():
    mqtt = MQTTClient('tester')

    mqtt.set_config({
        'reconnect_retries': UNLIMITED_RECONNECTS,
        'reconnect_delay': 1
    })

    await mqtt.connect('localhost')

    while True:
        #if mqtt.is_connected:
        mqtt.publish('ohlc_1m', 'check')
        await asyncio.sleep(1)

def handle_exception(loop, context):
    # context["message"] will always be there; but context["exception"] may not
    msg = context.get("exception", context["message"])
    logging.error(f"Caught exception: {str(msg)}")

loop = asyncio.get_event_loop()
loop.set_exception_handler(handle_exception)
loop.run_until_complete(main())

To test the reconnection I'm restarting mosquitto service and get the following output:

DEBUG:asyncio:Using selector: EpollSelector
INFO:gmqtt.mqtt.protocol:[CONNECTION MADE]
DEBUG:gmqtt.mqtt.handler:[CMD 0x20] b'\x00\x00\x03"\x00\n'
DEBUG:gmqtt.mqtt.handler:[CONNACK] flags: 0x0, result: 0x0
DEBUG:gmqtt.client:[QoS query IS EMPTY]
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.protocol:[RECV EMPTY] Connection will be reset automatically.
INFO:gmqtt.mqtt.protocol:[CONN CLOSE NORMALLY]
DEBUG:gmqtt.mqtt.handler:[CMD 0xe0] b''
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
Traceback (most recent call last):
  File "test.py", line 32, in <module>
    loop.run_until_complete(main())
  File "/usr/lib/python3.8/asyncio/base_events.py", line 612, in run_until_complete
    return future.result()
  File "test.py", line 22, in main
    mqtt.publish('ohlc_1m', 'check')
  File "/usr/lib/python3.8/site-packages/gmqtt/client.py", line 228, in publish
    mid, package = self._connection.publish(message)
  File "/usr/lib/python3.8/site-packages/gmqtt/mqtt/connection.py", line 54, in publish
    return self._protocol.send_publish(message)
  File "/usr/lib/python3.8/site-packages/gmqtt/mqtt/protocol.py", line 111, in send_publish
    self.write_data(pkg)
  File "/usr/lib/python3.8/site-packages/gmqtt/mqtt/protocol.py", line 45, in write_data
    if not self._transport.is_closing():
AttributeError: 'NoneType' object has no attribute 'is_closing'

I've seen a similar error in #74 so reporting this one for further investigation.

[naquad]

P. S. The same issue happens with 0.6.0

[naquad]

Here's what I've found out:

# gmqtt/mqtt/protocol.py:44
# method BaseMQTTProtocol::write_data

    def write_data(self, data: bytes):
        if not self._transport.is_closing(): ## Being throwed here as self._transport is None
            self._transport.write(data)
        else:
            logger.warning('[TRYING WRITE TO CLOSED SOCKET]')

The reason is that asyncio.StreamReaderProtocol::connection_lost method does self._transport = None in Python 3.8.

It is not clear for me if this problem should be handled somehow at the protocol handler layer or should be propagated to Client (QoS, packet retention and etc...).

[Lenka42]

@naquad could you please check 0.6.1?

[naquad]

@Lenka42 works! 🎉 Thank you 😄

If needed then here's the log output:

DEBUG:asyncio:Using selector: EpollSelector
INFO:gmqtt.mqtt.protocol:[CONNECTION MADE]
DEBUG:gmqtt.mqtt.handler:[CMD 0x20] b'\x00\x00\x03"\x00\n'
DEBUG:gmqtt.mqtt.handler:[CONNACK] flags: 0x0, result: 0x0
DEBUG:gmqtt.client:[QoS query IS EMPTY]
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.protocol:[RECV EMPTY] Connection will be reset automatically.
INFO:gmqtt.mqtt.protocol:[CONN CLOSE NORMALLY]
DEBUG:gmqtt.mqtt.handler:[CMD 0xe0] b''
WARNING:gmqtt.mqtt.protocol:[TRYING WRITE TO CLOSED SOCKET]
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
WARNING:gmqtt.mqtt.protocol:[TRYING WRITE TO CLOSED SOCKET]
INFO:gmqtt.mqtt.protocol:[CONNECTION MADE]
DEBUG:gmqtt.mqtt.handler:[CMD 0x20] b'\x00\x00\x03"\x00\n'
DEBUG:gmqtt.mqtt.handler:[CONNACK] flags: 0x0, result: 0x0
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.client:[QoS query IS EMPTY]
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)
DEBUG:gmqtt.mqtt.package:Sending PUBLISH (q0), 'ohlc_1m', ... (5 bytes)