channels-mqtt-proxy

KO | EN

gitlite — search

#javascript #python #hacktoberfest #react #ai #typescript #llm #go #golang #android #machine-learning #rust #deep-learning #linux

channels-mqtt-proxy

★ 14 Open GitHub ↗

A Channels 4.0 compatible MQTT worker

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# channels-mqtt-proxy

A Channels 3/4 compatible MQTT worker

This worker is a standard Channels Consumer which contains a
fully async MQTT server allowing channels messages to be used to
publish, subscribe to and receive MQTT messages

The proxy understands mqtt.subscribe and mqtt.publish so you can
change the topics dynamically.

When an MQTT subscribe is performed it is done on behalf of a channels
group and all Channels Consumers in that group will receive mqtt
messages as mqtt.message

The overview is :

MQTT <> Channels-MQTT-Proxy (in runworker) <> Redis/Channels-layer <> ASGI applications (in Daphne/runserver) <> Websockets/HTTP <> Browser

## Installation

```bash
pip install chanmqttproxy
```

## Usage

In Channels the asgi application handles all types of connection
routing.  Websocket and http connections are listened for by a
suitable server (eg daphne run by manage.py runserver) which will
instantiate classes and run objects to handle them. The 'mqtt'
Consumer does not accept incoming connections, just channel
messages. So it must be started as a worker which handles the Channel
messages; the MQTT client connection is then created inside the
MqttConsumer worker when the first channel message is received.

## Setup

The code may look familiar if you've used the Channels
[Chat tutorial](https://channels.readthedocs.io/en/stable/tutorial/index.html) :)

In fact it will add the ability to monitor the chat on the chat/<room>
MQTT channel and messages sent to that channel will appear on all
clients.

In `site/asgi.py`:

```python
	from chanmqttproxy import MqttConsumer
	from channels.routing import ChannelNameRouter, ProtocolTypeRouter
	
	application = ProtocolTypeRouter({
		"channel": ChannelNameRouter({
			"mqtt": MqttConsumer.as_asgi()
		}),
		... # rest of http/websocket routes
	})
```
To define the MQTT broker, in `site/settings.py`:

```python
# Local mqtt settings
MQTT_HOST = "mqtt.example.com"
MQTT_USER = "mqtt-test"
MQTT_PASSWORD = "mqtt-test"
MQTT_VERSION = 311  # defaults to 50
```

At this point you have a working async Channels/MQTT bridge

To subscribe to a topic in your AsyncConsumer

```python
    async def connect(self):
        ... # existing group_add() calls

# Join mqtt group
        await self.channel_layer.group_add(
            "mqttgroup",
            self.channel_name
        )
        # Ensure MQTT messages come to the room
        # This simplistic approach subscribes the room every
        # time a websocket connects but that's OK
        await self.channel_layer.send(
            "mqtt",
            {
                "type": "mqtt_subscribe",
                "topic": f"chat/{self.room_name}",
                "group": "mqttgroup",
        })

await self.accept()  # existing accept() call
```
To handle messages from a topic in your AsyncConsumer

```python
    # Receive message from mqtt group and send to websocket
    async def mqtt_message(self, event):
        message = event['message']
        topic = messagep["topic"]
        payload = messagep["payload"]

# Send message to WebSocket
        await self.send(text_data=json.dumps({
            'message': payload
        }))
```
To publish messages to a topic in your AsyncConsumer

```python
   # Receive message from WebSocket
    async def receive(self, text_data):

... # existing group_send etc

# Publish on mqtt too
        await self.channel_layer.send(
            "mqtt",
            {
                "type": "mqtt_publish",
                "publish": {  # These form the kwargs for mqtt.publish
                    "topic":  f"chat/{self.room_name}_out",
                    "payload": message,
                    "qos": 2,
                    "retain": False,
                    }
        })
```

For debug logging I use this at the end of settings.py:

```python
import logging.config

LOGGING_CONFIG = None

logging.config.dictConfig({
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'console': {
            # exact format is not important, this is the minimum information
            'format': '%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'console',
        },
    },
    'loggers': {
    # root logger
        'chanmqttproxy': {
            'level': 'DEBUG',
            'handlers': ['console'],
        },
        'mysite': {
            'level': 'DEBUG',
            'handlers': ['console'],
        },
        'chat': {
            'level': 'DEBUG',
            'handlers': ['console'],
        },
    },
})
```

The complete code for the Channels Chat tutorial application (up to
part 3) with the channels-mqtt-proxy additions is here:
https://github.com/lbt/channels-mqtt-proxy/tree/main/examples

## Example Usage

Create and activate a suitable venv.

Enter the `examples/` directory.

You'll need a Redis instance for Channels to work sensibly and an MQTT
broker like mosquitto of course.

Setup Django/channels etc in your venv by running:

```
pip install ..[examples]
```

Then edit `mysite/mysite/settings.py` to point to your mqtt server and run;

```python
python3 mysite/manage.py migrate
python3 mysite/manage.py runworker mqtt &
python3 mysite/manage.py runserver
```

Navigate to http://127.0.0.1:8000/chat/ and use "lobby" for the room.

Use your mqtt listener to listen to the topic `chat/lobby_out` and
publish to the topic `chat/lobby`

Notice that if we had just used chat/<room> for both topics then when
the chat room sends a message and the proxy client publishes it to
the MQTT topic the message would appear twice in the room. This is
because even if you're the one that publishes a message, if you're
subscribed to the topic, you will receive it too.

If you make changes to the code note that the Channels runworker does
not auto-reload and will still hold old subscribe/group information.

# TODO/Issues

## Connect to MQTT at startup
The MqttConsumer is only instantiated after the first message arrives
rather than when the worker starts. This means it may not be connected
so the await self.mqttproxy.connected.wait() is required on every
subscribe/publish (which is not a lot of overhead but...

## Unsubscribing and no-more-clients
It's not clear how to issue an unsubscribe or deal with all clients
disconnecting. If this is done in the AsyncConsumer disconnect() then
it needs a client-count which is probably unreliable. Currently MQTT
messages will always be sent to the Channels group and it handles
member timeout as per
https://channels.readthedocs.io/en/stable/channel_layer_spec.html#capacity
and
https://channels.readthedocs.io/en/stable/channel_layer_spec.html#persistence

## retain'ed messages
On initial subscribtion all retained messages are dropped.  This is
not ideal when retained messages are used to indicate 'last known
state' for clients.

However this is an MQTT concept and doesn't carry over to Channels
unless we handle retention and continue to store messages for each
Channel client that subscribes - and then somehow only send retained
messages to that new client and not the existing clients who've seen
them once..

Ideally there would be a mechanism for the app to pre-subscribe and
send retained (and new) messages to code that could update the
'current state' model which would then be maintained and used to
initialise new browser clients.

## MqttConsumer in worker doesn't exit
There doesn't seem to be a way to tell the worker to exit on Ctrl-C if
we trap it to clean up the MQTT connections.  Also note that in some
situations the Ctrl-C fails. Eg if the broker doesn't support V5.0 and
fallback to V311 is underway.

# Thanks
Thanks to Gurtam for https://github.com/wialon/gmqtt which is a great asyncio
MQTT client that I use extensively in my HA systems and integrate with PyQt too.

The Channels tutorial was really helpful in understanding the concepts.

Also to Xavier Lesa for https://github.com/xavierlesa/channels-asgi-mqtt which
is based on the paho-mqtt synchronous library and inspired me to write this.

channels-mqtt-proxy

오늘의 추천 프로젝트

Was this content helpful?

Recent Feedback

channels-mqtt-proxy

오늘의 추천 프로젝트

Was this content helpful?

Recent Feedback

Download README