[Bug] ntex-mqtt server unable to manage connections under high load

[gmartin82]

Describe the bug

The changes made for #39 introduced a secondary issue whereby the ntex-mqtt server struggles to manage MQTT connections when messages are being forwarded from Zenoh to MQTT clients at a high frequency.

Specifically, the server can't establish new connections with MQTT clients and can't maintain existing connections to clients due to failure to handle keep-alive messages. However, data is successfully forwarded from Zenoh to connected MQTT clients until the point where these connections time out.

This problem was known during the review of #39 and it was decided that the fix could be accepted with this issue as a known issue. Under common use cases of the MQTT plugin, it isn't expected to be an issue.

To reproduce

1. Run the Zenoh router using the attached config.
2. Setup and run a Mosquitto MQTT server with the following config:

$ cat /etc/mosquitto/conf.d/mosquitto-bridge-zenoh.conf
# Allow anonymous connections from `mosquitto_sub` clients
allow_anonymous true

# Redirect logs to be available via: `journalctl -u mosquitto -f`
log_dest stdout

listener 1883 localhost

connection mosquitto_zenoh_bridge

# zenoh-plugin-mqtt is listening on port 1884
address localhost:1884
topic # both 0

# zenoh-plugin-mqtt does not support... whatever this option does
try_private false

# Remap mosquitto broker status information from default of:
# $SYS/broker/connection/<remote_clientid>/state which is not a valid 
# zenoh key
notification_topic mqtt/broker/connection/mosquitto_zenoh_bridge/state

3. Run the following client app:

#!/bin/env python3

import time
import json
import logging
import argparse
import zenoh

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('-v', '--verbose', action='count', default=0)
    args = parser.parse_args()
    logging.basicConfig(level=logging.WARNING - (10 * args.verbose))

    try:
        session = zenoh.open()
        pub = session.declare_publisher('foo/bar')

        while True:
            pub.put(json.dumps('simple message'))

    # zenoh.ZError is not accessible?
    except Exception:
        logging.exception('Waiting for router connection:')

4. Observe that new MQTT clients are unable to connect and existing clients (e.g. the mosquito broker) disconnect after a few minutes.

System info

Platform: Ubuntu 20.04 64bit