Eclipse Paho MQTT C 라이브러리: MQTTClient와 MQTTAsync, 콜백 스레드, 재접속, Last Will

libVDA5050++ 아래에는 무엇이 있나

VDA5050 로봇 통신 모듈을 C++로 만들 때 많이 쓰는 libVDA5050++는 MQTT를 직접 구현하지 않습니다. 의존성 설정(cmake/ensure-deps.cmake, v3.2.2 기준)을 보면 Eclipse Paho의 C++ 라이브러리 paho.mqtt.cpp를 1.5.0으로 받아 오고, 그 빌드 옵션에 PAHO_WITH_MQTT_C ON을 넣어 paho.mqtt.cpp 저장소에 서브모듈로 들어 있는 C 라이브러리 paho.mqtt.c(이 버전에서는 v1.3.14)까지 함께 빌드합니다. 그래서 MQTT 브로커와 실제로 통신하는 것은 맨 아래의 C 라이브러리입니다.

flowchart TB
  A["libVDA5050++<br/>(VDA5050 프로토콜 로직, MqttModule)"] --> B["paho.mqtt.cpp<br/>(mqtt::async_client)"]
  B --> C["paho.mqtt.c<br/>(MQTTAsync API, libpaho-mqtt3as)"]
  C --> D["TCP / TLS(OpenSSL) 소켓"]
  D --> E(("MQTT 브로커"))

paho.mqtt.cpp의 mqtt::async_client는 C 라이브러리의 비동기 API인 MQTTAsync를 감싼 클래스이고, SSL을 켜면 C 쪽 paho-mqtt3as 라이브러리를 링크합니다. 그래서 libVDA5050++의 MQTT 동작(언제 재접속하는지, 콜백이 어느 스레드에서 도는지, 끊긴 동안 발행하면 어떻게 되는지)은 결국 MQTTAsync의 동작입니다. 의존성 설정의 주석에는 paho.mqtt.cpp 1.5.2에서 MSVC로 빌드할 때 일부 심볼이 빠지는 버그가 있어 1.5.0에 고정했다는 설명도 있습니다. 이렇게 버전을 고정한 이유는 직접 업그레이드하기 전에 확인해야 할 정보입니다.

이 글은 그 C 라이브러리를 다룹니다. 2026년 10월 기준 paho.mqtt.c의 최신 릴리스는 v1.3.16이고, API 이름과 기본값은 v1.3.16 헤더(MQTTAsync.h, MQTTClient.h)를 기준으로 확인했습니다. MQTT 자체(토픽, QoS, retain, Last Will)의 개념은 MQTT 발행-구독 모델에, VDA5050 메시지의 규칙은 VDA5050 order 수명주기에 정리해 두었습니다.

MQTT 패킷에서 payload가 차지하는 부분

이 글과 Paho 문서에는 payload라는 말이 계속 나오므로 먼저 정리해 둡니다. MQTT에서 메시지 하나는 PUBLISH 패킷 하나이고, 패킷은 헤더와 payload로 나뉩니다.

• 고정 헤더: 패킷 종류(PUBLISH), QoS, retain 플래그, 중복 전송 표시(DUP), 남은 길이.
• 가변 헤더: 토픽 이름과, QoS 1·2일 때의 패킷 ID(MQTT 5라면 속성도).
• payload: 그 뒤에 오는 실제 내용 바이트. 애플리케이션이 보내려는 데이터 자체입니다.

택배로 비유하면 헤더는 송장(어디로, 어떤 방식으로)이고 payload는 상자 안의 물건입니다. 브로커와 Paho 라이브러리는 송장만 보고 배달할 뿐, payload의 내용은 해석하지 않습니다. 그래서 payload는 JSON이든 바이너리든 아무 바이트열이나 될 수 있고, 형식은 주고받는 양쪽이 약속으로 정합니다. VDA5050은 그 약속을 “UTF-8 JSON, 메시지 종류별 JSON 스키마”로 정해 둔 것입니다. 예를 들어 uagv/v2/RobotCompany/0001/state 토픽의 PUBLISH 패킷에서 토픽 이름은 가변 헤더에, {"headerId":12,"orderId":"TO-5521",...} 같은 state JSON 전체가 payload에 들어갑니다.

Paho C에서 payload는 메시지 구조체의 두 필드로 다룹니다.

MQTTAsync_message msg = MQTTAsync_message_initializer;
msg.payload    = buf;     /* void*: 보낼 바이트의 시작 주소 */
msg.payloadlen = len;     /* int: 바이트 수 */
msg.qos        = 0;
msg.retained   = 0;

payload가 char*가 아니라 void*이고 길이를 따로 받는 것은, 내용이 문자열이라는 보장이 없기 때문입니다. 받는 쪽에서도 마찬가지로 끝에 NUL 문자가 붙어 있지 않으므로 payloadlen만큼만 읽어야 합니다. 이 점은 아래 수신 콜백 절에서 다시 다룹니다.

알아 두면 좋은 성질이 몇 가지 더 있습니다.

• 최대 크기: MQTT 패킷의 남은 길이 필드는 최대 268,435,455바이트(약 256MB)를 표현할 수 있지만, 실제 한도는 브로커 설정이 정합니다. Mosquitto 2.x라면 max_packet_size(예전 설정 이름은 message_size_limit)로 제한할 수 있고, 한도를 넘는 패킷은 브로커가 받아 주지 않습니다. 경로가 아주 긴 order나 큰 지도 정보처럼 payload가 커지는 메시지는 이 한도를 먼저 확인해야 합니다. VDA5050이 지도 파일을 MQTT로 보내지 않고 downloadMap 액션으로 다운로드 링크만 넘기는 것도 같은 맥락입니다.
• 빈 payload와 retain: retain 플래그를 켜고 payload가 0바이트인 메시지를 보내면, 브로커는 그 토픽에 보관하던 retained 메시지를 삭제합니다. 테스트하다 남은 connection·factsheet의 retained 메시지를 지울 때 쓰는 방법이고(mosquitto_pub -t <토픽> -r -n), 반대로 실수로 빈 payload를 retain으로 보내면 관제 시스템이 읽어야 할 마지막 상태가 사라집니다.
• 인코딩: 토픽 이름은 MQTT 규격상 UTF-8 문자열이어야 하지만 payload에는 그런 제약이 없습니다. VDA5050처럼 JSON을 쓰기로 했다면 문자 인코딩(UTF-8)을 양쪽이 맞춰야 하고, 한국어 설명 문자열을 넣는 actionDescription 같은 필드에서 인코딩이 어긋나면 파싱 오류가 납니다.

라이브러리 4종과 빌드

paho.mqtt.c를 빌드하면 API 종류(동기/비동기)와 TLS 지원 여부에 따라 라이브러리가 넷으로 나옵니다.

라이브러리	API	TLS	CMake 타깃
libpaho-mqtt3c	MQTTClient (동기)	없음	eclipse-paho-mqtt-c::paho-mqtt3c
libpaho-mqtt3cs	MQTTClient (동기)	있음	eclipse-paho-mqtt-c::paho-mqtt3cs
libpaho-mqtt3a	MQTTAsync (비동기)	없음	eclipse-paho-mqtt-c::paho-mqtt3a
libpaho-mqtt3as	MQTTAsync (비동기)	있음	eclipse-paho-mqtt-c::paho-mqtt3as

헤더도 API마다 따로라서 동기 API는 MQTTClient.h, 비동기 API는 MQTTAsync.h를 씁니다. 한 프로그램에서 둘을 섞어 링크할 필요는 없습니다. TLS가 필요 없는 개발 환경이라도 운영에서 ssl://로 바꿀 가능성이 있다면 처음부터 s가 붙은 쪽을 링크해 두는 편이 낫습니다. TLS 없는 라이브러리에 ssl:// 주소를 넘기면 접속 단계에서 MQTTASYNC_SSL_NOT_SUPPORTED(-13) 오류가 납니다.

우분투에서는 sudo apt install libpaho-mqtt-dev로 배포판 패키지를 설치할 수 있지만, 배포판 버전은 대개 최신 릴리스보다 늦습니다. 최신 릴리스가 필요하면 소스에서 빌드합니다.

git clone https://github.com/eclipse-paho/paho.mqtt.c.git
cd paho.mqtt.c
git checkout v1.3.16
cmake -B build -DPAHO_WITH_SSL=ON -DPAHO_BUILD_STATIC=ON -DPAHO_ENABLE_TESTING=OFF
cmake --build build -j
sudo cmake --install build

TLS 지원, 정적·공유 라이브러리, 테스트 빌드 여부를 CMake 옵션으로 정합니다.

옵션	기본값	의미
PAHO_WITH_SSL	OFF	OpenSSL을 쓰는 3cs·3as 라이브러리도 빌드
PAHO_BUILD_SHARED / PAHO_BUILD_STATIC	ON / OFF	공유·정적 라이브러리 빌드 여부
PAHO_ENABLE_TESTING	ON	테스트 빌드. 라이브러리만 필요하면 끄는 편이 빠름
PAHO_HIGH_PERFORMANCE	OFF	트레이스와 힙 추적 코드를 빼고 빌드
PAHO_WITH_UNIX_SOCKETS	OFF	유닉스 도메인 소켓 지원(유닉스 계열에서만)

PAHO_HIGH_PERFORMANCE는 따로 기억해 둘 만합니다. 이 옵션을 켜면 아래 “디버깅” 절의 MQTT_C_CLIENT_TRACE 트레이스가 아예 빠지는데, libVDA5050++는 paho.mqtt.cpp를 받아 빌드할 때 이 옵션을 켭니다. 통합 중에 프로토콜 수준 로그가 필요해지면 이 차이 때문에 헤매기 쉽습니다.

설치한 라이브러리는 CMake 패키지로 찾습니다.

find_package(eclipse-paho-mqtt-c REQUIRED)
add_executable(agv_client agv_client.c)
target_link_libraries(agv_client PRIVATE eclipse-paho-mqtt-c::paho-mqtt3as)

동기 API: MQTTClient

MQTTClient는 “호출하면 결과가 나올 때까지 기다리는” 방식입니다. 접속, 발행 완료 대기가 모두 블로킹이라 흐름이 순서대로 읽힙니다.

#include <stdio.h>
#include <string.h>
#include "MQTTClient.h"

int main(void) {
    MQTTClient client;
    MQTTClient_create(&client, "tcp://localhost:1883", "test-publisher",
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);

    MQTTClient_connectOptions conn = MQTTClient_connectOptions_initializer;
    conn.keepAliveInterval = 15;
    conn.cleansession = 1;

    int rc = MQTTClient_connect(client, &conn);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "connect failed: %s\n", MQTTClient_strerror(rc));
        MQTTClient_destroy(&client);
        return 1;
    }

    const char* payload = "{\"actions\":[]}";
    MQTTClient_message msg = MQTTClient_message_initializer;
    msg.payload = (void*)payload;
    msg.payloadlen = (int)strlen(payload);
    msg.qos = 1;
    msg.retained = 0;

    MQTTClient_deliveryToken token;
    MQTTClient_publishMessage(client, "uagv/v2/RobotCompany/0001/instantActions", &msg, &token);
    MQTTClient_waitForCompletion(client, token, 5000L);   // QoS 1 ack를 최대 5초 기다림

    MQTTClient_disconnect(client, 1000);
    MQTTClient_destroy(&client);
    return 0;
}

MQTTClient는 두 가지 모드로 동작합니다. 콜백을 등록하지 않으면 동기 모드로, 애플리케이션 스레드 하나에서 모든 일이 일어납니다. 이때 메시지를 받으려면 MQTTClient_receive()를 호출하거나, keep-alive와 ack 처리를 위해 MQTTClient_yield()를 주기적으로 불러야 합니다. 오래 아무것도 호출하지 않으면 keep-alive ping이 나가지 않아 브로커가 연결을 끊습니다. MQTTClient_setCallbacks()로 수신 콜백을 등록하면 비동기 모드가 되어 라이브러리가 백그라운드 스레드에서 수신과 ack를 처리하고 콜백을 그 스레드에서 호출합니다.

이름 때문에 헷갈리지만, MQTTClient의 “비동기 모드”는 수신만 백그라운드로 돌릴 뿐 접속과 발행 완료 대기는 여전히 블로킹입니다. 자동 재접속도 MQTTClient에는 없습니다. 연결이 끊기면 connectionLost 콜백이 불리고, 다시 접속하는 일은 애플리케이션 몫입니다. 테스트용 발행 도구나 짧은 스크립트에는 이 단순함이 장점이지만, 로봇처럼 하루 종일 떠 있으며 무선망 끊김을 견뎌야 하는 프로그램에는 다음 절의 MQTTAsync가 맞습니다.

비동기 API: MQTTAsync

MQTTAsync는 모든 호출이 즉시 반환합니다. MQTTAsync_connect()는 “접속을 시작하라”는 명령을 내부 큐에 넣고 바로 돌아오고, 결과는 나중에 onSuccess/onFailure 콜백으로 알려 줍니다. 발행과 구독도 같습니다. 라이브러리는 내부에 송신 스레드와 수신 스레드를 두고 큐의 명령을 처리하며, 콜백은 이 내부 스레드에서 호출됩니다.

이 글에서 쓰는 함수는 아래 표의 것들입니다.

함수	역할
MQTTAsync_create / MQTTAsync_createWithOptions	클라이언트 핸들 생성. 서버 주소, 클라이언트 ID, 영속성 방식 지정
MQTTAsync_setCallbacks	연결 끊김·메시지 수신·발행 완료 콜백 등록
MQTTAsync_setConnected	접속 성공 콜백 등록. 첫 접속과 자동 재접속 모두에서 호출됨
MQTTAsync_connect	접속 시작. 결과는 connect options의 onSuccess/onFailure로
MQTTAsync_subscribe	구독 시작. 결과는 response options의 콜백으로
MQTTAsync_sendMessage	발행 시작. 토픽과 payload는 라이브러리가 복사해 둠
MQTTAsync_waitForCompletion	특정 토큰의 완료를 기다림(종료 처리 등에서 사용)
MQTTAsync_isConnected	현재 접속 상태 확인
MQTTAsync_disconnect / MQTTAsync_destroy	접속 종료와 핸들 해제
MQTTAsync_strerror	오류 코드를 문자열로

MQTTAsync_sendMessage가 토픽과 payload를 복사한다는 점(소스의 MQTTAsync_send가 malloc·memcpy로 사본을 만들어 큐에 넣습니다) 덕분에, 지역 변수 버퍼에 JSON을 만들어 넘기고 함수를 빠져나가도 안전합니다.

VDA5050 로봇 클라이언트의 뼈대

아래는 VDA5050 2.x 규칙을 따르는 로봇 쪽 클라이언트의 최소 뼈대입니다. 접속할 때 connection 토픽에 Last Will(CONNECTIONBROKEN)을 걸고, 접속에 성공할 때마다 ONLINE을 발행하고 order·instantActions를 구독하며, 받은 메시지는 복사해서 다른 스레드로 넘깁니다. 메시지의 headerId·timestamp 같은 헤더 필드는 지면 관계로 줄였습니다.

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"

#define BROKER   "tcp://broker.local:1883"
#define CLIENTID "RobotCompany_0001"
#define PREFIX   "uagv/v2/RobotCompany/0001"

static MQTTAsync client;

/* 실제 구현에서는 락으로 보호한 큐에 넣고 제어 스레드가 꺼내 처리한다 */
void enqueue_incoming(const char* topic, const char* json, size_t len);

static void publish_connection(const char* state) {
    char payload[256];
    snprintf(payload, sizeof payload,
             "{\"manufacturer\":\"RobotCompany\",\"serialNumber\":\"0001\","
             "\"connectionState\":\"%s\"}", state);

    MQTTAsync_message msg = MQTTAsync_message_initializer;
    msg.payload = payload;
    msg.payloadlen = (int)strlen(payload);
    msg.qos = 1;          /* connection 토픽은 QoS 1 */
    msg.retained = 1;     /* 그리고 retain */

    int rc = MQTTAsync_sendMessage(client, PREFIX "/connection", &msg, NULL);
    if (rc != MQTTASYNC_SUCCESS)
        fprintf(stderr, "publish connection failed: %s\n", MQTTAsync_strerror(rc));
}

/* 첫 접속에도, 자동 재접속에도 호출된다 */
static void on_connected(void* ctx, char* cause) {
    (void)ctx; (void)cause;
    publish_connection("ONLINE");
    MQTTAsync_subscribe(client, PREFIX "/order", 0, NULL);
    MQTTAsync_subscribe(client, PREFIX "/instantActions", 0, NULL);
}

static void on_connection_lost(void* ctx, char* cause) {
    (void)ctx;
    fprintf(stderr, "connection lost: %s\n", cause ? cause : "unknown");
    /* automaticReconnect가 켜져 있으면 라이브러리가 다시 접속을 시도한다 */
}

static int on_message(void* ctx, char* topic, int topicLen, MQTTAsync_message* m) {
    (void)ctx; (void)topicLen;
    /* payload는 NUL로 끝나지 않는다. 길이와 함께 복사해서 넘긴다 */
    enqueue_incoming(topic, (const char*)m->payload, (size_t)m->payloadlen);
    MQTTAsync_freeMessage(&m);
    MQTTAsync_free(topic);
    return 1;   /* 처리 완료. 0을 반환하면 라이브러리가 다시 전달을 시도한다 */
}

static void on_connect_failure(void* ctx, MQTTAsync_failureData* resp) {
    (void)ctx;
    fprintf(stderr, "connect failed: rc=%d %s\n",
            resp ? resp->code : 0, (resp && resp->message) ? resp->message : "");
}

int mqtt_start(void) {
    int rc = MQTTAsync_create(&client, BROKER, CLIENTID,
                              MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTASYNC_SUCCESS) return rc;

    MQTTAsync_setCallbacks(client, NULL, on_connection_lost, on_message, NULL);
    MQTTAsync_setConnected(client, NULL, on_connected);

    static const char will_payload[] =
        "{\"manufacturer\":\"RobotCompany\",\"serialNumber\":\"0001\","
        "\"connectionState\":\"CONNECTIONBROKEN\"}";
    MQTTAsync_willOptions will = MQTTAsync_willOptions_initializer;
    will.topicName = PREFIX "/connection";
    will.message = will_payload;
    will.qos = 1;
    will.retained = 1;

    MQTTAsync_connectOptions conn = MQTTAsync_connectOptions_initializer;
    conn.keepAliveInterval = 15;     /* VDA5050 권장 heartbeat 간격 */
    conn.cleansession = 1;
    conn.MQTTVersion = MQTTVERSION_3_1_1;
    conn.automaticReconnect = 1;
    conn.minRetryInterval = 2;       /* 초 단위, 실패할 때마다 두 배 */
    conn.maxRetryInterval = 16;
    conn.will = &will;
    conn.onFailure = on_connect_failure;

    return MQTTAsync_connect(client, &conn);
}

void mqtt_stop(void) {
    publish_connection("OFFLINE");   /* 정상 종료는 Last Will이 나가지 않으므로 직접 보낸다 */

    MQTTAsync_disconnectOptions dis = MQTTAsync_disconnectOptions_initializer;
    dis.timeout = 2000;              /* 남은 발행을 최대 2초 동안 마저 보낸다 */
    MQTTAsync_disconnect(client, &dis);
    /* 실제 코드에서는 dis.onSuccess로 완료를 확인한 뒤 destroy 한다 */
    MQTTAsync_destroy(&client);
}

이 뼈대에서 VDA5050 통신 모듈과 직결되는 결정이 몇 가지 있습니다.

구독을 접속 성공 콜백에서 합니다. cleansession = 1이면 브로커는 연결이 끊길 때 구독 정보를 지웁니다. 자동 재접속에 성공해도 구독은 사라진 상태이므로, 재접속할 때마다 다시 구독해야 합니다. MQTTAsync_setConnected로 등록한 콜백은 소스상 첫 접속(onSuccess 호출 직후)과 자동 재접속 성공 모두에서 불리므로, 구독을 여기에 두면 두 경우를 한 곳에서 처리할 수 있습니다. 구독을 MQTTAsync_connect의 onSuccess에만 넣는 실수가 흔한데, 그러면 첫 접속 때는 잘 되다가 무선망이 한 번 끊겼다 붙은 뒤부터 로봇이 order를 하나도 받지 못합니다. 연결 상태 표시는 정상이라 원인을 찾기가 더 어렵습니다.

ONLINE도 접속 성공 콜백에서 다시 보냅니다. 연결이 비정상으로 끊기면 브로커가 Last Will로 CONNECTIONBROKEN을 retain해 두므로, 재접속한 뒤 ONLINE을 다시 보내지 않으면 관제 시스템은 로봇이 계속 끊겨 있다고 봅니다.

수신 콜백에서는 복사만 하고 돌아옵니다. 콜백은 라이브러리 내부 스레드에서 불리므로, 여기서 JSON을 길게 검증하거나 주행 제어 함수를 부르며 오래 붙잡고 있으면 그동안 다른 메시지 전달과 발행 결과 통지가 밀립니다. 받은 바이트를 큐에 넣고 바로 1을 반환한 뒤, 처리는 애플리케이션 스레드에서 하는 구조가 안전합니다.

콜백에서 지켜야 할 규칙

헤더 문서가 명시하는 규칙과, 어겼을 때 나타나는 증상입니다.

• 메시지와 토픽은 직접 해제합니다. 수신 콜백이 1을 반환할 때는 MQTTAsync_freeMessage(&message)와 MQTTAsync_free(topicName)를 호출해야 합니다. 빠뜨리면 메시지마다 메모리가 새고, order처럼 큰 JSON을 받는 로봇은 며칠 운영하면 눈에 띄게 메모리가 늘어납니다.
• 0을 반환하면 같은 메시지가 다시 옵니다. 0은 “아직 처리하지 못했다”는 뜻이라 라이브러리가 콜백을 다시 호출합니다. 이때는 메시지를 해제하면 안 됩니다(재전달할 때 해제된 메모리를 읽게 됩니다). 파싱 실패 같은 영구적인 오류에 0을 반환하면 같은 메시지가 끝없이 재전달되므로, 처리할 수 없는 메시지도 기록만 남기고 1을 반환하는 것이 맞습니다.
• payload는 문자열이 아닙니다. NUL 종료가 없고 길이는 payloadlen에 있습니다.
• 콜백 안에서 MQTTAsync_create·MQTTAsync_destroy를 부르지 않습니다. 헤더에 명시된 금지 사항입니다. 연결을 완전히 새로 만들어야 한다면 플래그만 세우고 다른 스레드에서 처리합니다.
• 콜백 안에서 발행·구독 호출은 괜찮습니다. 위 예제처럼 접속 성공 콜백에서 구독하는 것은 정상적인 사용법입니다. 다만 콜백 안에서 MQTTAsync_waitForCompletion으로 완료를 기다리면, 완료를 처리해야 할 스레드를 스스로 막는 셈이 되므로 피합니다.

C++ 래퍼를 쓸 때도 같습니다. paho.mqtt.cpp의 mqtt::callback을 상속해 구현하는 connected(), connection_lost(), message_arrived(), delivery_complete()는 위 C 콜백을 그대로 감싼 것이고, 같은 내부 스레드에서 호출됩니다. 메모리 해제는 래퍼가 해 주지만 “콜백에서 오래 붙잡지 말 것”은 그대로 적용됩니다.

접속 옵션의 기본값과 VDA5050에 맞춘 값

MQTTAsync_connectOptions_initializer의 기본값과 VDA5050 로봇 클라이언트에서 흔히 바꾸는 값은 다음과 같습니다(v1.3.16 헤더 기준).

필드	기본값	VDA5050 로봇에서	비고
keepAliveInterval	60초	15초 안팎	스펙이 끊김 감지 heartbeat를 15초 정도로 권장. 너무 짧으면 무선망 순간 끊김에도 Last Will이 나감
cleansession	1	1 또는 0	아래 “세션 유지” 참고
MQTTVersion	MQTTVERSION_DEFAULT	MQTTVERSION_3_1_1 명시	기본값은 3.1.1을 시도하고 실패하면 3.1로 내려감
connectTimeout	30초	환경에 맞게	접속 시도 한 번의 제한 시간
automaticReconnect	0 (꺼짐)	1	꺼져 있으면 끊긴 뒤 재접속은 애플리케이션 몫
minRetryInterval / maxRetryInterval	1초 / 60초	2초 / 16초 등	재시도 간격은 실패할 때마다 두 배, 최대값에서 멈춤
maxInflight	65535	기본값	QoS 1/2에서 ack를 기다리는 메시지 최대 수
will	NULL	connection 토픽 Last Will	QoS 1, retain, CONNECTIONBROKEN
ssl	NULL	운영 환경에서는 설정	ssl:// 주소와 함께 사용

재접속 간격을 정할 때는 로봇이 많은 현장을 떠올려야 합니다. 브로커를 재시작하면 수십 대 로봇이 동시에 끊기고, 모두 같은 간격으로 재접속을 시도하면 브로커가 다시 몰리는 접속에 시달립니다. 최소 간격을 너무 짧게 잡지 말고, 필요하면 로봇마다 시작 시점을 조금씩 다르게 두는 편이 좋습니다.

Last Will 메시지는 접속할 때 브로커에 미리 맡겨 두는 것이라, 그 안의 timestamp와 headerId는 항상 접속 시점의 값입니다. VDA5050 스펙도 이 점을 인정하고 있으므로 관제 시스템이 Last Will의 timestamp를 실제 끊긴 시각으로 해석하지 않게 해야 합니다.

세션 유지: cleansession = 0이 해결해 주지 않는 것

cleansession = 0으로 접속하면 브로커가 클라이언트 ID별로 세션(구독 정보, 전달하지 못한 메시지)을 유지합니다. “그럼 끊긴 동안 온 order도 재접속하면 받겠구나” 하고 기대하기 쉬운데, VDA5050 환경에서는 대개 그렇지 않습니다. VDA5050은 order와 instantActions에 QoS 0을 쓰도록 정하고, MQTT 3.1.1 스펙에서 브로커가 오프라인 클라이언트를 위해 보관해야 하는 것은 QoS 1·2 메시지이며 QoS 0 메시지 보관은 선택 사항입니다. Mosquitto도 기본 설정에서는 QoS 0 메시지를 쌓아 두지 않습니다.

그래서 세션 유지는 “재구독을 안 해도 된다”는 정도의 효과에 그치고, 끊긴 동안의 order는 관제 시스템이 state를 보고 다시 보내는 VDA5050의 재전송 구조에 맡기는 것이 맞습니다. 세션을 유지하면 클라이언트 ID가 겹칠 때 문제가 더 커진다는 점도 주의해야 합니다. 같은 ID로 두 프로세스가 접속하면 브로커는 먼저 접속한 쪽을 끊어 버리므로, 시뮬레이터와 실제 로봇이 같은 ID를 쓰면 둘이 번갈아 끊기며 Last Will이 계속 발행됩니다.

끊긴 동안 발행하면 어떻게 되나

로봇 제어 루프는 연결 상태와 관계없이 state를 만들어 발행하려 합니다. 이때 MQTTAsync의 동작은 생성 옵션(MQTTAsync_createOptions)에 달려 있습니다.

필드	기본값	의미
sendWhileDisconnected	0	0이면 끊긴 동안 MQTTAsync_sendMessage가 MQTTASYNC_DISCONNECTED(-3)를 반환
maxBufferedMessages	100	끊긴 동안 쌓아 둘 최대 메시지 수
deleteOldestMessages	0	버퍼가 차면 0은 새 메시지를, 1은 가장 오래된 메시지를 버림
allowDisconnectedSendAtAnyTime	0	한 번도 접속하지 않은 상태에서도 버퍼링 허용

기본값(MQTTAsync_create로 만든 경우)은 버퍼링을 하지 않으므로, 끊긴 동안의 발행은 오류 코드를 받고 버려집니다. 오류를 무시하고 넘어가도 되는 상황이지만, 반환값을 확인하지 않는 코드라면 “보낸 줄 알았던 state가 없다”는 혼란이 생길 수 있으니 로그로 남겨 두는 것이 좋습니다.

버퍼링을 켜면(MQTTAsync_createWithOptions에 sendWhileDisconnected = 1) 재접속 직후 쌓여 있던 메시지가 한꺼번에 나갑니다. VDA5050 state에는 이것이 반갑지 않습니다. 관제 시스템은 몇십 초 전의 위치와 진행 상황을 담은 state를 순서대로 받으며 로봇이 거꾸로 움직이는 듯한 화면을 보게 되고, 그 사이 order 판단을 잘못할 수도 있습니다. state는 “지금 상태”만 의미가 있으므로, 버퍼링하지 않고 접속 성공 콜백에서 현재 state를 새로 한 번 발행하는 쪽이 VDA5050의 의도에 맞습니다. 반대로 꼭 전달되어야 하는 기록성 메시지(현장 정책에 따라 남기는 이벤트 로그 등)를 같은 클라이언트로 보낸다면, 그 메시지만 별도로 저장했다가 재전송하는 편이 버퍼 옵션 하나로 모든 메시지를 쌓는 것보다 안전합니다.

영속성(persistence)

MQTTAsync_create의 네 번째 인자는 QoS 1·2 메시지의 전달 중 상태를 어디에 보관할지 정합니다.

값	의미
MQTTCLIENT_PERSISTENCE_NONE	메모리에만 보관. 프로세스가 죽으면 전달 중이던 메시지 정보도 사라짐
MQTTCLIENT_PERSISTENCE_DEFAULT	파일로 보관. 경로 인자가 NULL이면 작업 디렉터리에 클라이언트별 폴더를 만듦
MQTTCLIENT_PERSISTENCE_USER	애플리케이션이 저장 함수를 직접 제공

VDA5050 로봇 클라이언트는 대부분의 메시지가 QoS 0이라 영속성이 할 일이 거의 없으므로 NONE이 무난합니다. DEFAULT를 쓰면 서비스로 실행할 때 작업 디렉터리가 어디인지에 따라 엉뚱한 곳에 폴더가 생기거나, 읽기 전용 파일 시스템에서 생성 단계부터 MQTTASYNC_PERSISTENCE_ERROR(-2)가 날 수 있습니다. 파일 영속성을 켜 두면 재시작할 때 이전에 남은 메시지를 복원하려 하므로, MQTT 버전을 바꾸거나(3.1.1에서 5로) 클라이언트 ID 체계를 바꿀 때는 그 폴더를 정리해야 합니다.

TLS

브로커 주소를 ssl://broker.local:8883처럼 주고 MQTTAsync_SSLOptions를 connect options의 ssl에 넣습니다. WebSocket이 필요하면 ws://, wss:// 주소도 받습니다.

MQTTAsync_SSLOptions ssl = MQTTAsync_SSLOptions_initializer;
ssl.trustStore = "/etc/agv/certs/ca.crt";        /* 브로커 인증서를 검증할 CA */
ssl.keyStore = "/etc/agv/certs/robot-0001.crt";  /* 상호 인증(mTLS)일 때 클라이언트 인증서 */
ssl.privateKey = "/etc/agv/certs/robot-0001.key";
ssl.enableServerCertAuth = 1;                    /* 서버 인증서 검증 */
ssl.verify = 1;                                  /* 인증서의 호스트 이름 검증 */
conn.ssl = &ssl;

enableServerCertAuth와 verify는 역할이 다릅니다. 앞의 것은 인증서가 신뢰하는 CA로 서명됐는지, 뒤의 것은 인증서의 이름이 접속한 호스트 이름과 맞는지를 봅니다. 개발 중 자체 서명 인증서 때문에 둘 다 꺼 두었다가 그대로 운영에 나가는 경우가 흔하므로, 설정 파일에서 이 값이 환경별로 어떻게 들어가는지 확인해 두어야 합니다. 브로커 주소를 IP로 쓰는데 인증서에는 호스트 이름만 들어 있으면 verify = 1에서 핸드셰이크가 실패합니다. 이 경우 검증을 끄지 말고, 인증서의 SAN에 IP를 넣거나 주소를 호스트 이름으로 바꾸는 것이 맞습니다.

libVDA5050++ 설정이 C 옵션으로 이어지는 방식

libVDA5050++의 MQTT 설정(vda5050pp::config::MqttOptions)은 MqttModule(src/vda5050++/core/messages/mqtt_module.cpp)에서 paho.mqtt.cpp 옵션으로 옮겨지고, 그것이 다시 위의 C 구조체가 됩니다. v3.2.2 소스 기준으로 설정 이름이 어떻게 이어지는지 표로 옮겼습니다.

libVDA5050++ 설정	paho.mqtt.cpp	Paho C에서 대응하는 것
server	async_client(server, client_id)	MQTTAsync_create의 serverURI
(고정) client ID libvda5050++(agv_id=…)	같은 생성자	clientId
username, password	set_user_name, set_password	username, password
(고정) MQTT 3.1.1	set_mqtt_version(4)	MQTTVersion = 4
(고정) 세션 유지	set_clean_session(false)	cleansession = 0
min_retry_interval_, max_retry_interval_	set_automatic_reconnect(min, max) (기본 2초, 16초)	automaticReconnect, min/maxRetryInterval
keep_alive_interval_	set_keep_alive_interval (지정하지 않으면 라이브러리 기본값)	keepAliveInterval
connect_timeout_	set_connect_timeout	connectTimeout
use_ssl, enable_cert_check	ssl_options의 set_verify, set_enable_server_cert_auth	MQTTAsync_SSLOptions의 verify, enableServerCertAuth
(고정) Last Will	will_options (connection 토픽, QoS 1, retain, CONNECTIONBROKEN)	MQTTAsync_willOptions
interface, version_overwrite	토픽 접두어 {interface}/{version} (버전 기본값 v2)	(토픽 문자열)

소스를 보면 앞 절의 설계가 그대로 들어 있습니다. 구독은 connected() 콜백에서 QoS 0으로 하고, 같은 콜백에서 ONLINE을 보냅니다. 수신 콜백 message_arrived()에서는 토픽으로 order와 instantActions를 구분해 JSON을 파싱한 뒤 내부 이벤트로 넘기고, 구독하지 않은 토픽의 메시지가 오면 구독을 해제합니다. 정상 종료할 때는 OFFLINE을 보낸 뒤 disconnect()->wait()로 끝을 기다리고, factsheet는 retain으로 발행합니다.

여기서 운영할 때 알아 둘 점이 세 가지 있습니다.

1. keep-alive를 지정하지 않으면 60초입니다. libVDA5050++는 keep_alive_interval_을 설정했을 때만 값을 바꾸므로, 비워 두면 Paho 기본값이 적용됩니다. VDA5050이 권장하는 15초 안팎으로 끊김을 감지하려면 설정 파일에 명시해야 합니다.
2. 클라이언트 ID에 괄호와 =가 들어갑니다. MQTT 3.1.1은 브로커가 영숫자 1~23자 ID를 반드시 받도록 하고 그 밖의 ID는 받을지 말지를 브로커에 맡깁니다. Mosquitto처럼 대부분의 브로커는 문제없지만, ID 형식을 제한하는 클라우드 브로커나 ACL을 클라이언트 ID 패턴으로 거는 환경에서는 접속이 거부될 수 있습니다. 또 agv_id가 같은 설정을 두 프로세스가 쓰면 위에서 설명한 클라이언트 ID 충돌이 생기므로, 시뮬레이터용 설정을 복사할 때 반드시 바꿔야 합니다.
3. 트레이스가 빠진 빌드입니다. CPM으로 paho.mqtt.cpp를 받아 빌드할 때 PAHO_HIGH_PERFORMANCE ON을 쓰므로 MQTT_C_CLIENT_TRACE가 동작하지 않습니다. 시스템에 이미 설치된 Paho를 찾으면(find_package(PahoMqttCpp)) 그 빌드를 쓰므로, 트레이스가 필요하면 옵션을 끄고 빌드한 Paho를 먼저 설치해 두는 방법도 있습니다.

디버깅

트레이스를 지원하는 빌드라면 환경 변수만으로 프로토콜 수준 로그를 볼 수 있습니다.

export MQTT_C_CLIENT_TRACE=ON              # 표준 출력으로. 파일 경로를 주면 그 파일로
export MQTT_C_CLIENT_TRACE_LEVEL=PROTOCOL  # MQTT 패킷 송수신 수준. 더 자세히는 MAXIMUM 등
./agv_client

PROTOCOL 수준이면 CONNECT/CONNACK, SUBSCRIBE/SUBACK, PUBLISH, PINGREQ/PINGRESP가 시간과 함께 찍혀서, “구독은 했는가”, “ping이 제때 나가는가”, “브로커가 접속을 거부한 코드가 무엇인가”를 바로 확인할 수 있습니다.

API가 돌려주는 오류 코드는 MQTTAsync_strerror()로 문자열을 얻을 수 있습니다. 자주 보는 코드와 흔한 원인을 표로 모았습니다.

코드	이름	흔한 원인
-1	MQTTASYNC_FAILURE	일반 실패. failureData의 message와 함께 확인
-2	MQTTASYNC_PERSISTENCE_ERROR	파일 영속성 폴더를 만들거나 쓸 수 없음
-3	MQTTASYNC_DISCONNECTED	끊긴 상태에서 발행·구독(버퍼링 꺼짐)
-4	MQTTASYNC_MAX_MESSAGES_INFLIGHT	ack를 기다리는 메시지가 maxInflight를 넘음
-9	MQTTASYNC_BAD_QOS	QoS 값이 0~2 범위 밖
-12	MQTTASYNC_MAX_BUFFERED_MESSAGES	끊긴 동안 버퍼가 maxBufferedMessages에 도달
-13	MQTTASYNC_SSL_NOT_SUPPORTED	TLS 없는 라이브러리(3a)에 ssl:// 주소
-14	MQTTASYNC_BAD_PROTOCOL	주소 앞부분이 tcp://, ssl://, ws://, wss:// 등이 아님

트레이스를 쓸 수 없는 빌드에서는 브로커 쪽에서 보는 것이 가장 빠릅니다. 개발 브로커에서 mosquitto_sub -v -t 'uagv/v2/#'로 모든 메시지를 보고, Mosquitto 로그(log_type all)로 접속·끊김과 Last Will 발행 시점을 확인하면 클라이언트 쪽 로그 없이도 대부분의 연결 문제를 좁힐 수 있습니다.

자주 생기는 문제

증상	흔한 원인
첫 접속 때는 order를 받다가 무선망이 끊겼다 붙은 뒤로 못 받음	구독을 connect onSuccess에만 둠. 접속 성공 콜백(setConnected)에서 재구독
재접속 뒤 관제 화면에 계속 연결 끊김으로 표시	재접속 후 ONLINE을 다시 발행하지 않아 retain된 CONNECTIONBROKEN이 남음
로봇 두 대가 번갈아 끊김을 반복	같은 클라이언트 ID로 두 프로세스가 접속
수신 JSON 끝에 쓰레기 문자가 붙거나 파서가 가끔 실패	payload를 NUL 종료 문자열로 취급
같은 order가 계속 반복 처리됨	수신 콜백이 파싱 실패 등에서 0을 반환
며칠 운영하면 메모리가 계속 늘어남	수신 콜백에서 MQTTAsync_freeMessage·MQTTAsync_free 누락
재접속 직후 오래된 state가 한꺼번에 나감	sendWhileDisconnected 버퍼링을 켬
끊김 감지가 1분 넘게 걸림	keep-alive가 기본값 60초 그대로
서비스로 실행하면 생성 단계에서 실패	파일 영속성이 작업 디렉터리에 폴더를 만들지 못함
MQTT_C_CLIENT_TRACE를 켜도 로그가 없음	PAHO_HIGH_PERFORMANCE로 빌드된 라이브러리

참고 자료

• eclipse-paho/paho.mqtt.c (GitHub)
• eclipse-paho/paho.mqtt.cpp (GitHub)
• libVDA5050++ (Open Logistics Foundation)
• MQTT 3.1.1 명세 (OASIS)
• VDA5050 공식 저장소

같이 보면 좋은 글

• VDA5050 order 수명주기: 노드·엣지 시퀀스, base와 horizon, order update 이어붙이기
• VDA5050으로 AGV·AMR 관제 통신하기: MQTT 토픽, order·state·instantActions, libVDA5050++ C++ 구현
• MQTT 프로토콜과 발행-구독 메시징 모델 이해하기
• AGV·AMR 플릿 게이트웨이 개발 가이드 - 이기종 로봇과 관제 시스템을 잇는 미들웨어 아키텍처