MQTT PubSub

목표

로컬 MQTT 브로커를 실행합니다.
시뮬레이터 하나에서 MQTT로 게시하고 되돌려 받습니다.
게시자와 구독자 프로세스가 같은 토픽(topic)으로 메시지를 주고받습니다.

준비물

실습 환경의 apps 패키지와 환경 변수 OPCUA_APPS
MQTT 5를 지원하는 브로커. 이 페이지는 Eclipse Mosquitto 2.1.2로 실행했습니다.
터미널 세 개

시뮬레이터의 MQTT 전송은 MQTT 5와 QoS 1(at least once)을 씁니다. 메시지 본문은 UDP와 같은 UADP NetworkMessage입니다.

1. 브로커 실행

첫 번째 터미널에서 Mosquitto를 1883 포트로 실행합니다.

mosquitto -p 1883

설정 파일 없이 실행한 Mosquitto 2는 이 컴퓨터에서 오는 연결만 받고 익명 연결을 허용합니다. 출력에 mosquitto version 2.1.2 running이 보이면 준비된 것입니다. 이미 1883 포트를 쓰는 브로커가 있다면 그 브로커를 써도 됩니다.

2. 한 프로세스에서 게시하고 받기

두 번째 터미널에서 시뮬레이터를 실행합니다.

"$OPCUA_APPS/bin/pdcomm-opcua-headless"

전송을 mqtt로 바꿉니다. 브로커 기본값은 127.0.0.1:1883, 토픽 기본값은 pdcomm/opcua/simulator입니다.

{"action":"configure","config":{"pubsub":{"transport":"mqtt","broker":"127.0.0.1:1883"}}}

{"action":"pubsub_start"}

{"ok":true,"result":{"acknowledged":0,"destination":null,"dropped":0,"error":null,"listen":null,"queued":0,"received":0,"rejected":0,"role":"both","running":true,"sent":0,"transport":"mqtt"}}

both 역할은 토픽을 구독하고 같은 토픽으로 게시합니다.

{"action":"pubsub_publish"}

{"ok":true,"result":{"bytes":60,"fields":4,"sequence":1}}

{"action":"pubsub_status"}

{"ok":true,"result":{"acknowledged":1,"destination":null,"dropped":0,"error":null,"listen":null,"queued":1,"received":1,"rejected":0,"role":"both","running":true,"sent":1,"transport":"mqtt"}}

acknowledged는 브로커가 QoS 1 수신 확인(PUBACK)을 보낸 메시지 수입니다. 브로커를 거쳐 되돌아온 메시지를 꺼냅니다.

{"action":"pubsub_received"}

{"ok":true,"result":{"messages":[{"bytes":60,"datasets":[{"fields":[{"type":"Boolean","value":true},{"type":"Int32","value":42},{"type":"Double","value":3.5},{"type":"String","value":"pdcomm"}],"major":1,"minor":1,"sequence":1}],"publisher_id":77,"source":"mqtt","writer_group":1,"writer_ids":[17]}]}}

source가 mqtt이면 브로커에서 받은 메시지입니다. 프로그램을 끝냅니다.

{"action":"shutdown"}

3. 구독자와 게시자 나누기

두 번째 터미널에서 시뮬레이터를 다시 실행하고 구독자로 설정합니다. 토픽은 plant/line1/opcua입니다.

"$OPCUA_APPS/bin/pdcomm-opcua-headless"

{"action":"configure","config":{"pubsub":{"transport":"mqtt","role":"subscriber","broker":"127.0.0.1:1883","topic":"plant/line1/opcua"}}}

{"action":"pubsub_start"}

{"ok":true,"result":{"acknowledged":0,"destination":null,"dropped":0,"error":null,"listen":null,"queued":0,"received":0,"rejected":0,"role":"subscriber","running":true,"sent":0,"transport":"mqtt"}}

세 번째 터미널에서 시뮬레이터를 하나 더 실행하고 게시자로 설정합니다.

"$OPCUA_APPS/bin/pdcomm-opcua-headless"

{"action":"configure","config":{"pubsub":{"transport":"mqtt","role":"publisher","broker":"127.0.0.1:1883","topic":"plant/line1/opcua"}}}

{"action":"pubsub_start"}

{"action":"pubsub_publish","fields":[{"type":"Double","value":21.5},{"type":"String","value":"line-1"}]}

{"ok":true,"result":{"bytes":53,"fields":2,"sequence":1}}

{"action":"pubsub_status"}

{"ok":true,"result":{"acknowledged":1,"destination":null,"dropped":0,"error":null,"listen":null,"queued":0,"received":0,"rejected":0,"role":"publisher","running":true,"sent":1,"transport":"mqtt"}}

4. 구독자에서 확인

두 번째 터미널에서 받은 메시지를 꺼냅니다.

{"action":"pubsub_received"}

{"ok":true,"result":{"messages":[{"bytes":53,"datasets":[{"fields":[{"type":"Double","value":21.5},{"type":"String","value":"line-1"}],"major":1,"minor":1,"sequence":1}],"publisher_id":77,"source":"mqtt","writer_group":1,"writer_ids":[17]}]}}

5. 정리

두 시뮬레이터에 {"action":"shutdown"}을 입력하고, 브로커 터미널에서 Ctrl+C를 누릅니다.

자주 생기는 오류

응답의 `error`	원인과 조치
`MQTT broker connection failed: MQTT transport I/O: ConnectionRefused`	브로커 주소에서 기다리는 브로커가 없습니다. 브로커를 실행하고 주소와 포트를 확인합니다.
`MQTT topic must be 1 to 256 bytes without wildcards`	게시 토픽에는 와일드카드(`#`, `+`)를 쓸 수 없습니다.
`MQTT broker address must be an IP socket address`	브로커 주소는 `127.0.0.1:1883`처럼 IP 주소와 포트로 씁니다. 호스트 이름은 받지 않습니다.

시뮬레이터의 MQTT 연결은 사용자 이름, 암호, TLS를 쓰지 않습니다. 인증이 필요한 브로커에는 연결할 수 없습니다. GUI에서는 PubSub 탭의 Transport를 UADP over MQTT로 바꿉니다.