mqtt_websocket_proxy.py

#!usr/bin/env python3
# -*- coding: utf-8 -*-

import asyncio
import logging
from asyncio_mqtt import Client, MqttError
import websockets
 
ws = ""
logger = logging.getLogger(__name__)

async def mqtt_wait():
    while True:
        try:
            logger.info("Connecting to MQTT")
            async with Client("192.168.0.100") as client:
                logger.info("Connection to MQTT open")
                async with client.unfiltered_messages() as messages:
                    await client.subscribe('topic03/fromDevice')
                    async for message in messages:
                        logger.info("Message %s %s", message.topic, message.payload.decode())
                        mes = str(message.payload.decode())
                        await ws.send("echo : " + mes)
                # await asyncio.sleep(2)
        except MqttError as e:
            logger.error("Connection to MQTT closed: " + str(e))
        except Exception:
            logger.exception("Connection to MQTT closed")
        await asyncio.sleep(3)

async def accept(websocket, path):
  global ws
  ws = websocket
  while True:
    data = await websocket.recv()
    print("receive : " + data)
    await websocket.send("echo : " + data)
    await asyncio.sleep(3)
 
async def websocket_wait():
    start_server = websockets.serve(accept, "192.168.0.100", 9998)
    asyncio.get_event_loop().run_until_complete(start_server)
    asyncio.get_event_loop().run_forever()
    
def main():
    logging.basicConfig(level=logging.DEBUG,
                        format="%(asctime)s %(levelname)s %(message)s",
                        datefmt="%Y-%m-%d %H:%M:%S")
    asyncio.run(asyncio.wait([mqtt_wait(), websocket_wait()]))

main()