MQTT(mosquitto) + React WebSocket 통신

🪴 MQTT

• Message Queuing Telemetery Transport
• 사물인터넷(IoT)을 위한 표준 메시징 프로토콜
  • ex. 집안의 스마트 기기들처럼 다양한 IoT 장치들이 서로 소통할 수 있도록 해주는 기술

• Publish / Subscribe 구조
• 디바이스의 리소스를 적게 사용하도록 설계되어 오버헤드가 적게 발생
  • ex. 데이터 외에도 함께 전송되는 부가 정보와 같은 불필요한 리소스 사용을 최소화하도록 설게되어 있음
• device와 cloud 간의 양방향 통신 지원
  • ex. 집에 있는 스마트 조명 기기가 클라우드 서버에 신호를 보내면, 서버가 다시 그 기기에게 명령을 내림
• 수백만 개의 IoT 장치와 연결하도록 확장 가능
• connection 생성 시간이 짧아서 불안정한 네트워크 구간에서도 사용 가능

---

mosquitto

MQTT 브로커 서버 종류는 매우 다양하고, 지원하는 기능들에 차이가 있다.
MQTT Broker 비교

그 중에서 구조, 설치, 사용이 간단하고 이식성이 좋음, 가벼움, 사용자 많음 의 장점들을 가지고 있는 mosquitto 를 구축하는 방법이다.

나는 docker 서비스로 mosquitto를 설치, 빌드했는데
실제로 설치 용량이 작아서 비교적 빠르고 간단하게 구축할 수 있었다.

docker로 mosquitto 서버 올리기

최종 폴더 구조

mosquitto
├── config
│    ├── mosquitto.conf
│    └── pwfile
│
├── data
│    └── mosquitto.db
│
├── log
└── docker-compose.yml

docker-compose.yml

version: "3.7"

services:
  mosquitto:
    image: eclipse-mosquitto:2.0.18
    container_name: mosquitto
    restart: always
    ports:
      - "1883:1883"
      - "9001:9001"
    volumes:
      - ./config:/mosquitto/config:rw
      - ./data:/mosquitto/data:rw
      - ./log:/mosquitto/log:rw

mosquitto.conf

allow_anonymous false
listener 1883
listener 9001
protocol websockets
persistence true
password_file /mosquitto/config/pwfile
persistence_file mosquitto.db
persistence_location /mosquitto/data/

사용자 계정(필요 시) 비밀번호 파일 생성

$ touch config/pwfile

실행 & 빌드

$ docker-compose up -d --build

유저 계정 생성

# login interactively into the mqtt container
$ docker exec -it mosquitto sh

# Create new password file and add user and it will prompt for password
$ mosquitto_passwd -c /mosquitto/config/pwfile user1

# Add additional users (remove the -c option) and it will prompt for password
$ mosquitto_passwd /mosquitto/config/pwfile user2

# delete user command format
$ mosquitto_passwd -D /mosquitto/config/pwfile <user-name-to-delete>

# type 'exit' to exit out of docker container prompt

pwfile 파일을 열어보면 비밀번호 해시값이 저장되어 있는 것을 확인할 수 있음

user1:$7$101$dxSs5DOWacg2drF1$R1zwwZPf9qysg3IxG1kHdAHuOAOJFlODLngFRCfK8d084MhC2aSfy8CGCk4jsymwtGEy8H5IJwy9g/36kiegcw==

재실행하여 계정 적용

$ docker restart mosquitto

---

MQTT Explorer

메시지 발행/구독 테스트를 해볼 수 있음

download

MQTT Explorer

conection

---

React mqtt.js

MQTT.js 라이브러리를 사용해서 구독/발행 기능 구현

설치

$ npm install mqtt --save
# or
$ yarn add mqtt

Connecting

let mqttWS = useRef();

const connectMqtt = (serverUrl, userInfo) => {
  mqttWS.current = mqtt.connect(serverUrl, {
    // keepalive: 60,
    clientId: `web_${Math.random().toString(16).substr(2, 8)}`,
    ...userInfo
  });

  mqttWS.current.on("connect", () => {
    console.log("Connected");
  });

  mqttWS.current.on("error", (err) => {
    console.error("Connection error: ", err);
    mqttWS.current.end();
  });

  mqttWS.current.on("reconnect", () => {
    console.log("Reconnecting");
  });

  mqttWS.current.on("offline", () => {
    console.log("Went offline");
  });

  mqttWS.current.on("close", () => {
    console.log("Connection closed");
  });
};


useEffect(() => {
  connectMqtt(MQTT_URL, { username: "user1", password: "1234" });
  
  return () => {
    if (mqttWS.current)
      mqttWS.current.end(() => {
        console.log("Disconnected");
      });
  }
}, [])

Subscribe

// 커넥팅된 mqttWS ref 를 props로 받아서 활용

const subscribeMqtt = () => {
  mqttWS.current.subscribe(`TOPIC/${streamId}`, 0, (error) => {
    if (error) {
      console.log("Subscribe to topics error", error);
      return;
    } else {
      console.log(`Subscribe ${streamId}`);
    }
  });
  
  mqttWS.current.on("message", (topic, message) => {
    const payload = { topic, message: JSON.parse(message) };

    if (payload.message) {
      if (payload.topic === `TOPIC/${streamId}`) {
        const { bAlarmOccur, draw_info } = payload.message;
        setDrawInfo(draw_info);
      }
    } else {
      alert(JSON.parse(payload.message).error);
    }
  });
};


useEffect(() => {
  subscribeMqtt();
  
  return () => {
    if (mqttWS.current)
      mqttWS.current.unsubscribe(`TOPIC/${streamId}`, (error) => {
        if (error) {
          console.log("Unsubscribe error", error);
        } else {
          console.log(`Unsubscribe ${streamId}`);
        }
      });
  }
}, [])

Unsubscribe

mqttWS.current.unsubscribe(`TOPIC/${streamId}`, (error) => {
  if (error) {
    console.log("Unsubscribe error", error);
  } else {
    console.log(`Unsubscribe ${streamId}`);
  }
});

Disconnecting

mqttWS.current.end(() => {
  console.log("Disconnected");
});

실시간 영상 분석 컴포넌트에 적용한 모습

• 실시간 영상: MSE
• 분석 결과: MQTT