Fixed critical bugs e.g., incorrectly retained messages on qos0, race conditions on detached sessions, eliminated set_exception on client disconnect tasks, a few debug log message isEnabledFor wrappers.

hbmqtt/broker.py

@@ -707,21 +707,25 @@ class Broker:
                             if 'qos' in broadcast:
                                 qos = broadcast['qos']
                             if target_session.transitions.state == 'connected':
-                                self.logger.debug("broadcasting application message from %s on topic '%s' to %s" %
-                                                  (format_client_message(session=broadcast['session']),
-                                                   broadcast['topic'], format_client_message(session=target_session)))
+                                if self.logger.isEnabledFor(logging.DEBUG):
+                                    self.logger.debug("broadcasting application message from %s on topic '%s' to %s" %
+                                                      (format_client_message(session=broadcast['session']),
+                                                       broadcast['topic'], format_client_message(session=target_session)))
                                 handler = self._get_handler(target_session)
                                 task = asyncio.ensure_future(
                                     handler.mqtt_publish(broadcast['topic'], broadcast['data'], qos, retain=False),
                                     loop=self._loop)
                                 running_tasks.append(task)
-                            else:
-                                self.logger.debug("retaining application message from %s on topic '%s' to client '%s'" %
-                                                  (format_client_message(session=broadcast['session']),
-                                                   broadcast['topic'], format_client_message(session=target_session)))
+                            elif qos is not None and qos > 0:
+                                if self.logger.isEnabledFor(logging.DEBUG):
+                                    self.logger.debug("retaining application message from %s on topic '%s' to client '%s'" %
+                                                      (format_client_message(session=broadcast['session']),
+                                                       broadcast['topic'], format_client_message(session=target_session)))
                                 retained_message = RetainedApplicationMessage(
                                     broadcast['session'], broadcast['topic'], broadcast['data'], qos)
                                 yield from target_session.retained_messages.put(retained_message)
+                                if self.logger.isEnabledFor(logging.DEBUG):
+                                    self.logger.debug(f'target_session.retained_messages={target_session.retained_messages.qsize()}')
         except CancelledError:
             # Wait until current broadcasting tasks end
             if running_tasks:

hbmqtt/client.py

@@ -456,7 +456,7 @@ class MQTTClient:
             while self.client_tasks:
                 task = self.client_tasks.popleft()
                 if not task.done():
-                    task.set_exception(ClientException("Connection lost"))
+                    task.cancel()
 
         self.logger.debug("Watch broker disconnection")
         # Wait for disconnection from broker (like connection lost)

hbmqtt/mqtt/protocol/handler.py

@@ -417,7 +417,7 @@ class ProtocolHandler:
                         if task:
                             running_tasks.append(task)
                 else:
-                    self.logger.debug("%s No more data (EOF received), stopping reader coro" % self.session.client_id)
+                    self.logger.debug("No more data (EOF received), stopping reader coro")
                     break
             except MQTTException:
                 self.logger.debug("Message discarded")
@@ -425,10 +425,10 @@ class ProtocolHandler:
                 self.logger.debug("Task cancelled, reader loop ending")
                 break
             except asyncio.TimeoutError:
-                self.logger.debug("%s Input stream read timeout" % self.session.client_id)
+                self.logger.debug("Input stream read timeout")
                 self.handle_read_timeout()
             except NoDataException:
-                self.logger.debug("%s No data available" % self.session.client_id)
+                self.logger.debug("No data available")
             except BaseException as e:
                 self.logger.warning("%s Unhandled exception in reader coro: %r" % (type(self).__name__, e))
                 break
@@ -436,7 +436,7 @@ class ProtocolHandler:
             running_tasks.popleft().cancel()
         yield from self.handle_connection_closed()
         self._reader_stopped.set()
-        self.logger.debug("%s Reader coro stopped" % self.session.client_id)
+        self.logger.debug("Reader coro stopped")
         yield from self.stop()
 
     @asyncio.coroutine
@@ -457,6 +457,8 @@ class ProtocolHandler:
 
     @asyncio.coroutine
     def mqtt_deliver_next_message(self):
+        if not self._is_attached():
+            return None
         if self.logger.isEnabledFor(logging.DEBUG):
             self.logger.debug("%d message(s) available for delivery" % self.session.delivered_message_queue.qsize())
         try:

hbmqtt/plugins/manager.py

@@ -149,7 +149,8 @@ class PluginManager:
         if wait:
             if tasks:
                 yield from asyncio.wait(tasks, loop=self._loop)
-        self.logger.debug("Plugins len(_fired_events)=%d" % (len(self._fired_events)))
+        if self.logger.isEnabledFor(logging.DEBUG):
+            self.logger.debug("Plugins len(_fired_events)=%d" % (len(self._fired_events)))
 
     @asyncio.coroutine
     def map(self, coro, *args, **kwargs):