{"action":"created","issue":{"url":"https://api.github.com/repos/beerfactory/hbmqtt/issues/94","repository_url":"https://api.github.com/repos/beerfactory/hbmqtt","labels_url":"https://api.github.com/repos/beerfactory/hbmqtt/issues/94/labels{/name}","comments_url":"https://api.github.com/repos/beerfactory/hbmqtt/issues/94/comments","events_url":"https://api.github.com/repos/beerfactory/hbmqtt/issues/94/events","html_url":"https://github.com/beerfactory/hbmqtt/issues/94","id":269354149,"number":94,"title":"Properly cancelling task during pending MQTT subscription","user":{"login":"vlcinsky","id":635911,"avatar_url":"https://avatars3.githubusercontent.com/u/635911?v=4","gravatar_id":"","url":"https://api.github.com/users/vlcinsky","html_url":"https://github.com/vlcinsky","followers_url":"https://api.github.com/users/vlcinsky/followers","following_url":"https://api.github.com/users/vlcinsky/following{/other_user}","gists_url":"https://api.github.com/users/vlcinsky/gists{/gist_id}","starred_url":"https://api.github.com/users/vlcinsky/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vlcinsky/subscriptions","organizations_url":"https://api.github.com/users/vlcinsky/orgs","repos_url":"https://api.github.com/users/vlcinsky/repos","events_url":"https://api.github.com/users/vlcinsky/events{/privacy}","received_events_url":"https://api.github.com/users/vlcinsky/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2017-10-28T22:13:44Z","updated_at":"2018-03-17T11:56:43Z","closed_at":null,"author_association":"CONTRIBUTOR","body":"I am trying to create a task, which consumes data from MQTT topics and does some processing. As this is long term running process, it runs in `while True` loop and I want to cancel the task properly by sending `task.cancel()`.\r\n\r\nHowever, the `cancel()` results in exceptions thrown from within `RuntimeError: 'Event loop is closed.')\"`\r\n\r\nHere is the code which is adapted from project examples:\r\n\r\n\r\n\timport logging\r\n\timport asyncio\r\n\r\n\tfrom hbmqtt.client import MQTTClient, ClientException\r\n\tfrom hbmqtt.mqtt.constants import QOS_1, QOS_2\r\n\r\n\tlogger = logging.getLogger(__file__)\r\n\tlogging.basicConfig(level=logging.DEBUG)\r\n\r\n\r\n\tasync def reader_coro(mqtt_url, loop):\r\n\t    C = MQTTClient(loop=loop)\r\n\t    try:\r\n\t\tawait C.connect(mqtt_url)\r\n\t\tawait C.subscribe([\r\n\t\t\t('$SYS/broker/uptime', QOS_1),\r\n\t\t\t('$SYS/broker/load/#', QOS_2),\r\n\t\t    ])\r\n\t\twhile True:\r\n\t\t    message = await C.deliver_message()\r\n\t\t    packet = message.publish_packet\r\n\t\t    topic_name = packet.variable_header.topic_name\r\n\t\t    data = packet.payload.data\r\n\t\t    print(f\"TOPIC: {topic_name} MESSAGE: {data}\")\r\n\t    except asyncio.CancelledError:\r\n\t\tawait C.unsubscribe(['$SYS/broker/uptime', '$SYS/broker/load/#'])\r\n\t\tawait C.disconnect()\r\n\t\tlogger.debug(\"DISCONNECTED======================\")\r\n\t    except ClientException as ce:\r\n\t\tlogger.exception(\"Client exception: %s\" % ce)\r\n\r\n\r\n\tif __name__ == '__main__':\r\n\t    mqtt_url = 'mqtt://test.mosquitto.org/'\r\n\t    loop = asyncio.get_event_loop()\r\n\t    reader = loop.create_task(reader_coro(mqtt_url, loop))\r\n\t    try:\r\n\t\tloop.run_until_complete(reader)\r\n\t    except KeyboardInterrupt:\r\n\t\tlogger.debug(\"KEYBOARD INTERRUPT==================\")\r\n\t\treader.cancel()\r\n\t\tlogger.debug(\"READER CANCELLED==================\")\r\n\t\tloop.run_until_complete(reader)\r\n\t\tlogger.debug(\"FINISHED LOOP==================\")\r\n\t    finally:\r\n\t\tloop.close()\r\n\r\n\r\nWhen run, after Ctrl-C it prints out:\r\n\r\n\tDEBUG:hbmqtt.client:Waiting message delivery\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:0 message(s) available for delivery\r\n\t^CDEBUG:sub.py:KEYBOARD INTERRUPT==================\r\n\tDEBUG:sub.py:READER CANCELLED==================\r\n\tDEBUG:hbmqtt.client.plugins.packet_logger_plugin:hbmqtt/IyM^`78qi35Ns?ke -out-> UnsubscribePacket(ts=2017\r\n\t-10-29 00:11:05.438820, fixed=MQTTFixedHeader(length=42, flags=0x2), variable=PacketIdVariableHeader(pack\r\n\tet_id=2), payload=<hbmqtt.mqtt.unsubscribe.UnubscribePayload object at 0x7f40d879ea48>)\r\n\tDEBUG:hbmqtt.client.plugins.packet_logger_plugin:hbmqtt/IyM^`78qi35Ns?ke <-in-- UnsubackPacket(ts=2017-10\r\n\t-29 00:11:05.476021, fixed=MQTTFixedHeader(length=2, flags=0x0), variable=PacketIdVariableHeader(packet_i\r\n\td=2), payload=None)\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 puback waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 pucomp waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 purec waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 purel waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:waiting for tasks to be stopped\r\n\tDEBUG:hbmqtt.client.plugins.packet_logger_plugin:hbmqtt/IyM^`78qi35Ns?ke -out-> DisconnectPacket(ts=2017-\r\n\t10-29 00:11:05.477688, fixed=MQTTFixedHeader(length=0, flags=0x0), variable=None, payload=None)\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Task cancelled, reader loop ending\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Broker closed connection\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:hbmqtt/IyM^`78qi35Ns?ke Reader coro stopped\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 puback waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 pucomp waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 purec waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:Stopping 0 purel waiters\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:waiting for tasks to be stopped\r\n\tDEBUG:hbmqtt.mqtt.protocol.handler:closing writer\r\n\tDEBUG:transitions.core:Initiating transition from state connected to state disconnected...\r\n\tDEBUG:transitions.core:Exiting state connected. Processing callbacks...\r\n\tINFO:transitions.core:Exited state connected\r\n\tDEBUG:transitions.core:Entering state disconnected. Processing callbacks...\r\n\tINFO:transitions.core:Entered state disconnected\r\n\tDEBUG:sub.py:DISCONNECTED======================\r\n\tDEBUG:sub.py:FINISHED LOOP==================\r\n\tException ignored in: <generator object ProtocolHandler.mqtt_deliver_next_message at 0x7f40d87807d8>\r\n\tTraceback (most recent call last):\r\n\t  File \"/home/javl/devel/tis/dpo-tis-pns-pump/.tox/py36/lib/python3.6/site-packages/hbmqtt/mqtt/protocol/\r\n\thandler.py\", line 466, in mqtt_deliver_next_message\r\n\t    message = yield from self.session.delivered_message_queue.get()\r\n\t  File \"/home/javl/.pyenv/versions/3.6.1/lib/python3.6/asyncio/queues.py\", line 169, in get\r\n\t    getter.cancel()  # Just in case getter is not done yet.\r\n\t  File \"/home/javl/.pyenv/versions/3.6.1/lib/python3.6/asyncio/base_events.py\", line 573, in call_soon\r\n\t    self._check_closed()\r\n\t  File \"/home/javl/.pyenv/versions/3.6.1/lib/python3.6/asyncio/base_events.py\", line 357, in _check_close\r\n\td\r\n\t    raise RuntimeError('Event loop is closed')\r\n\tRuntimeError: Event loop is closed\r\n\tERROR:asyncio:Task was destroyed but it is pending!\r\n\ttask: <Task pending coro=<ProtocolHandler.mqtt_deliver_next_message() done, defined at /home/javl/devel/t\r\n\tis/dpo-tis-pns-pump/.tox/py36/lib/python3.6/site-packages/hbmqtt/mqtt/protocol/handler.py:461> wait_for=<\r\n\tFuture cancelled> cb=[_wait.<locals>._on_completion() at /home/javl/.pyenv/versions/3.6.1/lib/python3.6/a\r\n\tsyncio/tasks.py:374]>\r\n\t$\r\n\r\nIs there proper way of stopping pending subscription without such problems?\r\n"},"comment":{"url":"https://api.github.com/repos/beerfactory/hbmqtt/issues/comments/373914631","html_url":"https://github.com/beerfactory/hbmqtt/issues/94#issuecomment-373914631","issue_url":"https://api.github.com/repos/beerfactory/hbmqtt/issues/94","id":373914631,"user":{"login":"litnimax","id":1144539,"avatar_url":"https://avatars3.githubusercontent.com/u/1144539?v=4","gravatar_id":"","url":"https://api.github.com/users/litnimax","html_url":"https://github.com/litnimax","followers_url":"https://api.github.com/users/litnimax/followers","following_url":"https://api.github.com/users/litnimax/following{/other_user}","gists_url":"https://api.github.com/users/litnimax/gists{/gist_id}","starred_url":"https://api.github.com/users/litnimax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/litnimax/subscriptions","organizations_url":"https://api.github.com/users/litnimax/orgs","repos_url":"https://api.github.com/users/litnimax/repos","events_url":"https://api.github.com/users/litnimax/events{/privacy}","received_events_url":"https://api.github.com/users/litnimax/received_events","type":"User","site_admin":false},"created_at":"2018-03-17T11:56:43Z","updated_at":"2018-03-17T11:56:43Z","author_association":"NONE","body":"The problem is that it cannot be cancelled.\r\nPlease have a look.\r\nThis is what a do at __init__:\r\n```python\r\n        for signame in ('SIGINT', 'SIGTERM'):\r\n            self.loop.add_signal_handler(getattr(signal, signame),\r\n                lambda: asyncio.ensure_future(self.stop()))\r\n```\r\nAnd here is my stop function:\r\n```python\r\nasync def stop(self):\r\n        logger.info('Stopping mqttrpc...')\r\n        # Check subscriptions\r\n        await self.unsubscribe(self.subscriptions)\r\n        await self.disconnect()\r\n        tasks = [task for task in asyncio.Task.all_tasks() if task is not\r\n                    asyncio.tasks.Task.current_task()]\r\n        list(map(lambda task: task.cancel(), tasks))\r\n        results = await asyncio.gather(*tasks, return_exceptions=True)\r\n        logger.debug('Finished cancelling tasks, result: {}'.format(results))\r\n```\r\nAnd this hangs forever:\r\n```\r\n^C2018-03-17 13:56:24,098 - INFO - mqtt_rpc - Stopping mqttrpc...\r\n2018-03-17 13:56:24,098 - WARNING - hbmqtt.client - Client not connected, waiting for it\r\n2018-03-17 13:56:24,947 - WARNING - hbmqtt.client - MQTT connection failed: OSError(\"Multiple exceptions: [Errno 61] Connect call failed ('::1', 1883), [Errno 61] Connect call failed ('127.0.0.1', 1883)\",)\r\n2018-03-17 13:56:24,947 - DEBUG - transitions.core - Initiating transition from state disconnected to state disconnected...\r\n2018-03-17 13:56:24,947 - DEBUG - transitions.core - Exiting state disconnected. Processing callbacks...\r\n2018-03-17 13:56:24,947 - INFO - transitions.core - Exited state disconnected\r\n2018-03-17 13:56:24,947 - DEBUG - transitions.core - Entering state disconnected. Processing callbacks...\r\n2018-03-17 13:56:24,947 - INFO - transitions.core - Entered state\r\n2018-03-17 13:55:48,808 - WARNING - hbmqtt.client - Reconnection attempt failed: ConnectException(OSError(\"Multiple exceptions: [Errno 61] Connect call failed ('::1', 1883), [Errno 61] Connect call failed ('127.0.0.1', 1883)\",),)\r\n2018-03-17 13:55:50,814 - WARNING - hbmqtt.client - MQTT connection failed: OSError(\"Multiple exceptions: [Errno 61] Connect call failed ('::1', 1883), [Errno 61] Connect call failed ('127.0.0.1', 1883)\",)\r\n2018-03-17 13:55:50,814 - DEBUG - transitions.core - Initiating transition from state disconnected to state disconnected...\r\n2018-03-17 13:55:50,814 - DEBUG - transitions.core - Exiting state disconnected. Processing callbacks...\r\n2018-03-17 13:55:50,814 - INFO - transitions.core - Exited state disconnected\r\n2018-03-17 13:55:50,814 - DEBUG - transitions.core - Entering state disconnected. Processing callbacks...\r\n2018-03-17 13:55:50,815 - INFO - transitions.core - Entered state disconnected\r\n2018-03-17 13:55:50,815 - WARNING - hbmqtt.client - Reconnection attempt failed: ConnectException(OSError(\"Multiple exceptions: [Errno 61] Connect call failed ('::1', 1883), [Errno 61] Connect call failed ('127.0.0.1', 1883)\",),)\r\n```\r\nWhen I press CTRL+C I see immediately:\r\n```\r\n2018-03-17 13:56:24,098 - WARNING - hbmqtt.client - Client not connected, waiting for it\r\n```\r\nAnd it will not exit.\r\nPlease advise."}}
{"actor":{"display_login":"litnimax"}}