***********
MQTT client
***********
About
#####
Asynchronous MQTT client for 3.1.1 protocol version (mod).
Because of abandoned original repo this fork adds support for python >= 3.10
Installation
############
Recommended way (via pip):
.. code:: bash
$ pip install aio-mqtt
Example
#######
Simple echo server:
.. code:: python
import asyncio as aio
import logging
import typing as ty
import aio_mqtt
logger = logging.getLogger(__name__)
class EchoServer:
def __init__(
self,
reconnection_interval: int = 10,
loop: ty.Optional[aio.AbstractEventLoop] = None
) -> None:
self._reconnection_interval = reconnection_interval
self._loop = loop or aio.get_event_loop()
self._client = aio_mqtt.Client(loop=self._loop)
self._tasks = [
self._loop.create_task(self._connect_forever()),
self._loop.create_task(self._handle_messages())
]
async def close(self) -> None:
for task in self._tasks:
if task.done():
continue
task.cancel()
try:
await task
except aio.CancelledError:
pass
if self._client.is_connected():
await self._client.disconnect()
async def _handle_messages(self) -> None:
async for message in self._client.delivered_messages('in'):
while True:
try:
await self._client.publish(
aio_mqtt.PublishableMessage(
topic_name='out',
payload=message.payload,
qos=aio_mqtt.QOSLevel.QOS_1
)
)
except aio_mqtt.ConnectionClosedError as e:
logger.error("Connection closed", exc_info=e)
await self._client.wait_for_connect()
continue
except Exception as e:
logger.error("Unhandled exception during echo message publishing", exc_info=e)
break
async def _connect_forever(self) -> None:
while True:
try:
connect_result = await self._client.connect('localhost')
logger.info("Connected")
await self._client.subscribe(('in', aio_mqtt.QOSLevel.QOS_1))
logger.info("Wait for network interruptions...")
await connect_result.disconnect_reason
except aio.CancelledError:
raise
except aio_mqtt.AccessRefusedError as e:
logger.error("Access refused", exc_info=e)
except aio_mqtt.ConnectionLostError as e:
logger.error("Connection lost. Will retry in %d seconds", self._reconnection_interval, exc_info=e)
await aio.sleep(self._reconnection_interval, loop=self._loop)
except aio_mqtt.ConnectionCloseForcedError as e:
logger.error("Connection close forced", exc_info=e)
return
except Exception as e:
logger.error("Unhandled exception during connecting", exc_info=e)
return
else:
logger.info("Disconnected")
return
if __name__ == '__main__':
logging.basicConfig(
level='DEBUG'
)
loop = aio.new_event_loop()
server = EchoServer(reconnection_interval=10, loop=loop)
try:
loop.run_forever()
except KeyboardInterrupt:
pass
finally:
loop.run_until_complete(server.close())
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
License
#######
Copyright 2019-2020 Not Just A Toy Corp.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Raw data
{
"_id": null,
"home_page": "https://github.com/devbis/aio-mqtt",
"name": "aio-mqtt-mod",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.6.0",
"maintainer_email": "",
"keywords": "mqtt asyncio",
"author": "Not Just A Toy Corp.",
"author_email": "dev@notjustatoy.com",
"download_url": "https://files.pythonhosted.org/packages/af/19/ec24221723bddcd7ae28113d6ac9691f386dc50a3ed2767808658055adb8/aio-mqtt-mod-0.3.4.tar.gz",
"platform": null,
"description": "***********\nMQTT client\n***********\n\nAbout\n#####\n\nAsynchronous MQTT client for 3.1.1 protocol version (mod).\nBecause of abandoned original repo this fork adds support for python >= 3.10\n\nInstallation\n############\n\nRecommended way (via pip):\n\n.. code:: bash\n\n $ pip install aio-mqtt\n\nExample\n#######\n\nSimple echo server:\n\n.. code:: python\n\n import asyncio as aio\n import logging\n import typing as ty\n\n import aio_mqtt\n\n logger = logging.getLogger(__name__)\n\n\n class EchoServer:\n\n def __init__(\n self,\n reconnection_interval: int = 10,\n loop: ty.Optional[aio.AbstractEventLoop] = None\n ) -> None:\n self._reconnection_interval = reconnection_interval\n self._loop = loop or aio.get_event_loop()\n self._client = aio_mqtt.Client(loop=self._loop)\n self._tasks = [\n self._loop.create_task(self._connect_forever()),\n self._loop.create_task(self._handle_messages())\n ]\n\n async def close(self) -> None:\n for task in self._tasks:\n if task.done():\n continue\n task.cancel()\n try:\n await task\n except aio.CancelledError:\n pass\n if self._client.is_connected():\n await self._client.disconnect()\n\n async def _handle_messages(self) -> None:\n async for message in self._client.delivered_messages('in'):\n while True:\n try:\n await self._client.publish(\n aio_mqtt.PublishableMessage(\n topic_name='out',\n payload=message.payload,\n qos=aio_mqtt.QOSLevel.QOS_1\n )\n )\n except aio_mqtt.ConnectionClosedError as e:\n logger.error(\"Connection closed\", exc_info=e)\n await self._client.wait_for_connect()\n continue\n\n except Exception as e:\n logger.error(\"Unhandled exception during echo message publishing\", exc_info=e)\n\n break\n\n async def _connect_forever(self) -> None:\n while True:\n try:\n connect_result = await self._client.connect('localhost')\n logger.info(\"Connected\")\n\n await self._client.subscribe(('in', aio_mqtt.QOSLevel.QOS_1))\n\n logger.info(\"Wait for network interruptions...\")\n await connect_result.disconnect_reason\n except aio.CancelledError:\n raise\n\n except aio_mqtt.AccessRefusedError as e:\n logger.error(\"Access refused\", exc_info=e)\n\n except aio_mqtt.ConnectionLostError as e:\n logger.error(\"Connection lost. Will retry in %d seconds\", self._reconnection_interval, exc_info=e)\n await aio.sleep(self._reconnection_interval, loop=self._loop)\n\n except aio_mqtt.ConnectionCloseForcedError as e:\n logger.error(\"Connection close forced\", exc_info=e)\n return\n\n except Exception as e:\n logger.error(\"Unhandled exception during connecting\", exc_info=e)\n return\n\n else:\n logger.info(\"Disconnected\")\n return\n\n\n if __name__ == '__main__':\n logging.basicConfig(\n level='DEBUG'\n )\n loop = aio.new_event_loop()\n server = EchoServer(reconnection_interval=10, loop=loop)\n try:\n loop.run_forever()\n except KeyboardInterrupt:\n pass\n\n finally:\n loop.run_until_complete(server.close())\n loop.run_until_complete(loop.shutdown_asyncgens())\n loop.close()\n\nLicense\n#######\n\nCopyright 2019-2020 Not Just A Toy Corp.\n\nLicensed under the Apache License, Version 2.0 (the \"License\");\nyou may not use this file except in compliance with the License.\nYou may obtain a copy of the License at\n\n http://www.apache.org/licenses/LICENSE-2.0\n\nUnless required by applicable law or agreed to in writing, software\ndistributed under the License is distributed on an \"AS IS\" BASIS,\nWITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\nSee the License for the specific language governing permissions and\nlimitations under the License.\n",
"bugtrack_url": null,
"license": "Apache License 2.0",
"summary": "Asynchronous MQTT client for 3.1.1 protocol version.",
"version": "0.3.4",
"project_urls": {
"Homepage": "https://github.com/devbis/aio-mqtt"
},
"split_keywords": [
"mqtt",
"asyncio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "1fbddd304b995f43599eec7730e4b989803fb6bd01783dfc8a5aa0e726018640",
"md5": "4887abe91004bae55b77ef09f0778030",
"sha256": "598223aebe38b69780633b9c546d109e39964b5a605305af8b690b9f90624753"
},
"downloads": -1,
"filename": "aio_mqtt_mod-0.3.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "4887abe91004bae55b77ef09f0778030",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.6.0",
"size": 18416,
"upload_time": "2023-07-28T08:43:27",
"upload_time_iso_8601": "2023-07-28T08:43:27.281898Z",
"url": "https://files.pythonhosted.org/packages/1f/bd/dd304b995f43599eec7730e4b989803fb6bd01783dfc8a5aa0e726018640/aio_mqtt_mod-0.3.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "af19ec24221723bddcd7ae28113d6ac9691f386dc50a3ed2767808658055adb8",
"md5": "94330efe6ac946aefa7131737c2d675a",
"sha256": "340184b35771b7eb7982072fcca313213d856638dd7f98b99bda3ab16ba23552"
},
"downloads": -1,
"filename": "aio-mqtt-mod-0.3.4.tar.gz",
"has_sig": false,
"md5_digest": "94330efe6ac946aefa7131737c2d675a",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.6.0",
"size": 15624,
"upload_time": "2023-07-28T08:43:29",
"upload_time_iso_8601": "2023-07-28T08:43:29.170009Z",
"url": "https://files.pythonhosted.org/packages/af/19/ec24221723bddcd7ae28113d6ac9691f386dc50a3ed2767808658055adb8/aio-mqtt-mod-0.3.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-07-28 08:43:29",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "devbis",
"github_project": "aio-mqtt",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"tox": true,
"lcname": "aio-mqtt-mod"
}
JSON
