개요

MQTT Publish 데이터를 Flask에서 Subscribe 하여 받은 후, Vue로 socket 통신하는 과정을 기록하려 한다.

MQTT 통신은 IoT 기술에 많이 쓰이는데 현재 프로젝트에서도 센서 데이터를 무선으로 통신하기 위해 사용한다.

Communication Step

1. 센서 데이터 측정 (보드)
2. gateway 로 무선 통신 - TI 15.4 stack
3. gateway에서 MQTT Broker(서버)에 데이터 publish. - node.js
4. MQTT Broker에 publish한 topic subscribe. - flask
5. subscribe하여 받은 데이터 db 저장. - flask
6. socket emit으로 frontend로 통신 - flask -> vue.js
7. socket.on 으로 데이터 받기 - vue.js

센서 데이터를 측정하여 gateway로 보내고, gateway에서 센서 데이터를 MQTT Publish로 발행한다.

이를 서버(Flask)에서 MQTT Subscribe로 구독하여 데이터를 받고 DB에도 저장. 그 후, Socket emit을 통해 Vue.js로 보낸 후 UI 상에 실시간 데이터를 보여줄 수 있도록 한다.

MQTT Subscribe 과정 (Flask)

✅ 참고 사이트 : https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php#using-callback
이전 프로젝트에서는 publish에 paho.mqtt를 써서 이번에도 사용하였다.

결론적으로 이 모듈은 socket 통신과 함께 사용할 수 없어 다른 모듈을 사용했다.

먼저 paho.mqtt를 사용하여 삽질한 과정은 아래와 같다.

paho.mqtt.subscribe (실패)

Install Module

pip install paho-mqtt

Main Code

# backend/main.py

from flask import Flask
from flask_socketio import SocketIO
app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*", engineio_logger=False)

from backend.ws import *
socketio.run(app, debug=True, host="0.0.0.0", port=port)

Communication Code (Use callback)

# backend/ws.py
import paho.mqtt.subscribe as subscribe
from backend import app, socketio
from flask_socketio import emit

def on_message_print(client, userdata, message):
  try:
    print(message.payload)
    emit('data', message.payload)
  except Exception as e: 
    print(e)

@socketio.on('connect', namespace='/receiver')
def connect():
    print("Connected")
    subscribe.callback(on_message_print, "/example", hostname="localhost", port=1883)

@socketio.on('disconnect', namespace='/receiver')
def disconnect():
    print('Disconnected')
   

통신 실패

처음에는 paho.mqtt.subscribe 모듈을 사용하여 subscribe를 callback 하는 형식으로 socket 메세지를 받으려 했다.

callback으로 받은 message.payload를 출력하니 값이 제대로 받아지기에 이를 emit으로 socket 통신을 해보니 vue에서 받는 데이터가 전혀 없었다.
그래서 다른 방법인 paho.mqtt.client를 이용했다.

paho.mqtt.client (실패)

✅ 참고 사이트 : https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php#connect-reconnect-disconnect

Communication Code (Use loop_start)

# backend/ws.py
import paho.mqtt.client as client
from backend import app, socketio
from flask_socketio import emit

def on_connect(client, userdata, rc, data):
  print("connect")
  
def on_message(client, userdata, msg, data):
    print(msg.payload)
    emit('data', msg.payload

@socketio.on('connect', namespace='/receiver')
def connect():
    print("Connected")
    client = mqtt.Client()
    client.username_pw_set(username, password)
    client.on_connect = on_connect
    client.on_message = on_message
    client.subscribe('/example')
    client.connect('localhost', 1883)
    client.loop_start()

@socketio.on('disconnect', namespace='/receiver')
def disconnect():
    print('Disconnected')
    client.disconnect()

통신 실패

여전히 socket 통신이 되지 않았다.

RuntimeError: Working outside of request context.

이런 오류가 발생하여 그냥 emit이 아닌 SocketIO로 정의한 socketio.emit을 사용하였다. 에러는 뜨지 않았으나 여전히 통신이 되지 않았고, 이상하게 도중에 flask를 종료시키니 그 동안 쌓여있던 emit 메시지가 vue로 한 번에 날라가는 현상이 있었다.
loop_start 이후 mqtt 통신으로만 thread가 돌다가 종료되고 나서 쌓인 socket 통신이 뒤늦게 동작하는 것 같았다. thread를 적절히 사용하면 해결할 수 있을 것 같아 다양하게 수정해보았지만 Exception in thread Thread-1: 와 같은 오류로 Thread-1, 4, 5 까지 다양한 에러를 보았다.

원인 분석

이 paho.mqtt 모듈의 정확한 사용법과 해결방법을 발견하지 못했으나 개인적으로 생각해본건 socketio 를 선언할 때 Flask의 app을 연결하듯이 (socketio = SocketIO(app)) mqtt도 이러한 형태로 구현해야 socket과 mqtt가 동시에 통신할 수 있을 것 같다.

결론은 paho.mqtt를 사용했을 때는 socket 통신과 mqtt 통신이 동시에 되지 않는 것이다.

flask_mqtt

✅ 참고 사이트 : https://flask-mqtt.readthedocs.io/en/latest/usage.html#interact-with-socketio

Install Module

pip install Flask-MQTT

Main Code

# backend/main.py

from flask import Flask
from flask_socketio import SocketIO
from flask_mqtt import Mqtt

app = Flask(__name__)
app.config['MQTT_BROKER_URL'] = "localhost"
app.config['MQTT_BROKER_PORT'] = 1883
mqtt = Mqtt(app)
socketio = SocketIO(app, cors_allowed_origins="*", engineio_logger=False)

from backend.ws import *
socketio.run(app, debug=True, host="0.0.0.0", port=8080)

Communication Code

# backend/ws.py
from backend import app, socketio, mqtt
from flask_socketio import emit

@mqtt.on_connect()
def handle_connect(client, userdata, flags, rc):
    print("mqtt connect")

@mqtt.on_message()
def handle_mqtt_message(client, userdata, message):
  mqtt_data = json.loads(message.payload.decode())

  socketio.emit('data', mqtt_data, namespace='/receiver')
  
@socketio.on('connect', namespace='/receiver')
def connect():
    print("Connected")
    mqtt.subscribe('/example')

@socketio.on('disconnect', namespace='/receiver')
def disconnect():
    print('Disconnected')
    mqtt.unsubscribe('/example')

해결

이 모듈에서 방법을 찾았다.
mqtt와 socketio에 app을 연결시켜 통신하도록 하면 동시에 통신이 된다!!!!
이거 때문에 이틀 내내 골머리를 앓았는데 해결 돼서 정말 다행이다 😊💚🧡💛💙💜

혹시 나 처럼 flask에서 mqtt와 socket.io를 같이 쓸 일이 있다면 이걸 참고하자 ㅠㅠㅠㅠㅠㅠ