diff --git a/hbmqtt/broker.py b/hbmqtt/broker.py index 7bb8bd6..55beb95 100644 --- a/hbmqtt/broker.py +++ b/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: diff --git a/hbmqtt/client.py b/hbmqtt/client.py index 4df0526..6846878 100644 --- a/hbmqtt/client.py +++ b/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) diff --git a/hbmqtt/mqtt/protocol/handler.py b/hbmqtt/mqtt/protocol/handler.py index 3c294bc..f231a14 100644 --- a/hbmqtt/mqtt/protocol/handler.py +++ b/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: diff --git a/hbmqtt/plugins/manager.py b/hbmqtt/plugins/manager.py index 7a923a8..c96c5f1 100644 --- a/hbmqtt/plugins/manager.py +++ b/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):