반응 시스템용 MQTT 클라이언트 구현

MQTT-Reactive는 LiamBindle의 MQTT-C 라이브러리에서 파생된 MQTT v3.1.1 클라이언트입니다. MQTT-Reactive의 목표는 반응성 임베디드 시스템에서 사용하기 위해 C로 작성된 휴대용 및 비차단 MQTT 클라이언트를 제공하는 것입니다. 우선, 이 기사에서는 반응 시스템이 무엇인지 설명합니다. 그런 다음 이러한 종류의 시스템에 적합한 소프트웨어 구조를 설계하는 방법을 설명합니다. 마지막으로 이 기사에서는 상태 머신과 이벤트 기반 패러다임을 사용하여 반응 시스템에서 MQTT-Reactive 라이브러리를 사용하는 방법을 보여줍니다. 이를 위해 기사에서는 실제 IoT 장치를 상태 머신, 상호 작용 및 구조와 같은 UML 다이어그램을 사용하여 소프트웨어 구조 및 상태 기반 동작을 설명하는 실증적인 예로 사용합니다. 이 기사는 또한 IoT 장치의 MQTT-Reactive 클라이언트를 C 언어로 구현하기 위한 지침을 제공합니다.

많은 임베디드 시스템은 내부 또는 외부 이벤트에 반응합니다. 이러한 반응이 완료되면 소프트웨어는 다시 돌아가서 다음 이벤트를 기다립니다. 이것이 이벤트 구동 시스템을 반응 시스템이라고도 하는 이유입니다.

이벤트 기반 프로그래밍 또는 반응 프로그래밍은 반응 시스템을 위한 유연하고 예측 가능하며 유지 관리 가능한 소프트웨어를 구현하는 데 가장 적합한 프로그래밍 패러다임 중 하나입니다. 이 패러다임에서 프로그램의 흐름은 이벤트에 의해 결정됩니다. 종종 반응형 소프트웨어의 구조는 여러 종류의 이벤트를 기다리고 처리하는 활성 개체로 알려진 여러 동시 장치로 구성됩니다. 각 활성 개체는 들어오는 이벤트를 처리하는 제어 스레드와 이벤트 대기열을 소유합니다. 반응 시스템에서 활성 개체는 일반적으로 상태 차트에 정의된 상태 기반 동작을 갖습니다.

다중 동시 작업이 있는 반응 시스템에서 MQTT-Reactive 라이브러리를 사용하고 상태 머신과 이벤트 기반 패러다임을 모두 사용하는 방법을 탐색하기 위해 IoT 장치를 예로 사용합니다.

MQTT 프로토콜을 사용하는 아이디어는 철도 회사를 위해 IoT 장치를 개발하던 중에 태어났습니다. 이 장치는 다음을 수행할 수 있는 명확한 반응 시스템이었습니다.

<올>

여러 디지털 입력의 변경 감지 및 저장

여러 아날로그 신호 수집, 필터링 및 저장

저장된 정보를 원격 서버에 주기적으로 보내기

GSM 네트워크를 통해 MQTT 프로토콜을 통해 정보 보내기 및 받기

MQTT는 GSM 네트워크와 같이 대기 시간이 길고 데이터 속도가 낮은 링크가 예상되는 IoT 및 네트워킹 애플리케이션에서 일반적으로 사용되는 경량 게시자-가입자 기반 메시징 프로토콜이기 때문에 선택되었습니다.

언급된 IoT 장치에 대한 MQTT 기능은 수정된 버전의 LiamBindle MQTT-C를 사용하여 수행되었습니다. 해당 장치의 소프트웨어는 반응형 소프트웨어로 설계되었기 때문에 MQTT-C는 비동기식 이벤트를 교환하여 나머지 시스템과 통신하도록 수정되어야 했습니다. 이러한 이벤트는 네트워크를 통해 트래픽을 수신 및 전송하고 중요한 정보를 서버에 연결하고 게시하는 데 사용되었습니다. 결과 소프트웨어 라이브러리는 MQTT-Reactive라고 불렸습니다.

상태 머신

MQTT-Reactive는 MQTT-Reactive 클라이언트의 기본 동작을 모델링하는 그림 1과 같이 상태 머신을 통해 사용되었습니다. MqttMgr(MQTT Manager)이라는 활성 개체였습니다. 그림 1의 상태 머신 작업은 상태 머신에서 MQTT-Reactive 라이브러리를 사용하는 방법을 보여줍니다. 그림 1에서는 액션 언어로 C 언어를 사용했지만, 어떤 컴퓨터 언어나 형식 언어를 사용할 수 있습니다.

더 큰 이미지를 보려면 클릭하세요.

그림 1. MQTT-Reactive 클라이언트의 상태 머신(출처:VortexMakes)

그림 1의 상태 머신은 WaitingForNetConnection 상태에서 시작합니다. 네트워크 연결이 서버에 설정된 후 WaitingForNetConnection은 Activate 이벤트를 수신하고 상태 시스템은 WaitingForSync 상태로 전환됩니다. 이 상태에서만 상태 머신은 Connect 및 Publish 이벤트를 통해 각각 CONNECT 또는 PUBLISH와 같은 브로커에 보낼 MQTT 메시지를 준비할 수 있습니다. Sync 상태는 Sync 상태의 내부 구획에 포함된 defer 키워드로 지정된 Publish 이벤트를 지연하기 위해 UML의 특수 메커니즘을 사용합니다. 동기화가 현재 상태일 때 게시 이벤트가 발생하면 SM이 WaitingForSync 또는 WaitingForNetConnection과 같은 지연된 이벤트 목록에 게시 이벤트가 없는 상태가 될 때까지 향후 처리를 위해 저장(지연)됩니다. 이러한 상태에 진입하면 상태 시스템은 저장된 모든 게시 이벤트를 자동으로 호출한 다음 전환 대상 상태에 따라 이 이벤트를 사용하거나 삭제합니다.

SyncTime 밀리초마다 상태 시스템은 Receive 및 Send 이벤트를 네트워크 관리자에 게시하여 네트워크에서 트래픽을 실제로 보내고 받는 Sync 복합 상태로 전환합니다. 네트워크 문제를 처리하는 동시 개체입니다.

도입된 MqttMgr은 CONNECT 및 PUBLISH 패킷만 지원하지만 다소 간단한 변경으로 SUBSCRIBE 패킷을 지원할 수 있습니다.

상태 머신 작업은 params 키워드를 사용하여 소비된 이벤트의 매개변수에 액세스합니다. 예를 들어, 다음 전환에서 Connect 이벤트는 해당 MqttMgr 개체의 속성을 업데이트하는 데 사용되는 두 개의 매개변수 clientId 및 keepAlive를 전달합니다.

Connect(clientId, keepAlive)/ me->clientId =params->clientId; me->keepAlive =params->keepAlive; me->operRes =mqtt_connect(&me->client, me->clientId, NULL, NULL, 0, NULL, NULL, 0, me->keepAlive);

이 예에서 Connect(clientId, keepAlive) 이벤트는 전환의 트리거이고 mqtt_connect() 호출은 결과적으로 실행되는 작업의 일부입니다. 즉, MqttMgr 객체가 'publishing_client' 및 '400' 매개변수로 Connect(clientId, keepAlive) 이벤트를 수신하면 Connect("publishing_client", 400), MqttMgr의 clientId 및 keepAlive 속성이 ' 값으로 업데이트됩니다. 결과적으로 'publishing_client' 및 '400'입니다.

이벤트를 생성하고 보내기 위해 상태 머신의 동작은 GEN() 매크로를 사용합니다. 예를 들어 다음 명령문은 Collector 개체에 Receive 이벤트를 전송하며 이는 itsCollector 포인터에 의해 MqttMgr 개체의 속성으로 참조됩니다.

GEN(me->itsCollector, Receive());

GEN() 문의 첫 번째 인수는 이벤트를 수신하는 객체이고, 두 번째 인수는 이벤트 인수(있는 경우)를 포함하여 전송되는 이벤트입니다. 인수는 이벤트 매개변수와 일치해야 합니다. 예를 들어 다음 명령문은 ConnRefused(code) 이벤트를 생성하고 이를 이벤트 매개변수로 브로커가 반환한 코드를 전달하는 Collector 객체로 보냅니다.

GEN(me->itsCollector, ConRefused(코드));

params 키워드를 사용하여 소비된 이벤트의 매개변수에 액세스하고 GEN() 매크로를 사용하여 조치에서 이벤트를 생성하는 아이디어는 순전히 설명 목적으로 Rational Rhapsody Developer의 코드 생성기에서 채택되었습니다.

그림 1에서 상태 머신의 기본 동작은 브로커로부터 연결 수락을 수신할 때마다 MQTT-Reactive에 의해 호출되는 콜백을 설정합니다. 이 콜백은 MqttMgr 코드 내에서 구현되어야 합니다. 이 콜백은 아래와 같이 Collector 개체로 전송되기 위해 ConnAccepted 또는 ConnRefused(code) 이벤트를 생성해야 합니다.

정적  무효 connack_response_callback (열거 MQTTConnackReturnCode return_code){ /*...*/ if (return_code ==MQTT_CONNACK_ACCEPTED) { GEN(me->itsCollector, ConnAccepted()); } 기타 { GEN(me->itsCollector, ConnRefused(return_code)); }}

모델 구현

그림 1의 모델은 선호하는 소프트웨어 도구 또는 자체 상태 머신 구현을 사용하여 C 또는 C++로 구현할 수 있습니다. RKH 프레임워크, QP 프레임워크, Yakindu Statechart 도구 또는 Rational Rhapsody Developer와 같이 이를 수행하기 위해 인터넷에서 사용할 수 있는 다양한 도구가 있습니다. 모두 Statechart와 C/C++ 언어를 지원합니다. 또한 일부는 Statechart 다이어그램을 그리고 이 다이어그램에서 코드를 생성하는 도구를 포함합니다.

이 상태 머신은 MQTT-Reactive 코드의 엄격한 캡슐화를 제공하고 MQTT-Reactive 기능을 호출하거나 MQTT-Reactive 데이터에 액세스할 수 있는 유일한 엔티티인 MqttMgr(MQTT Manager)이라는 활성 개체에서 실행되었습니다. 시스템의 다른 동시 엔터티와 모든 ISR은 MqttMgr과 이벤트를 교환하여 간접적으로만 MQTT-Reactive를 사용할 수 있었습니다. 동시 엔터티를 동기화하고 데이터를 공유하기 위해 이 메커니즘을 사용하면 세마포어, 뮤텍스, 지연 또는 이벤트 플래그와 같은 기존 차단 메커니즘의 위험을 피할 수 있습니다. 이러한 메커니즘은 진단 및 수정이 어렵고 지루한 예기치 않은 오작동을 일으킬 수 있습니다.

MqttMgr 활성 개체는 속성을 데이터 항목 집합으로 캡슐화합니다. 데이터 항목은 이름과 유형이 있는 변수를 지정하며, 여기서 유형은 실제로 데이터 유형입니다. MqttMgr 개체의 데이터 항목은 개체 구조의 구성원에 매핑됩니다. 멤버의 이름과 타입은 객체 데이터와 동일합니다. 예를 들어, MqttMgr 개체 유형의 클라이언트 속성은 MqttMgr 구조 내부에 데이터 멤버로 값에 의해 포함됩니다.

구조체 MqttMgr { /* ... */ 구조체 mqtt_client 클라이언트; /* 속성 클라이언트 */ LocalRecvAll localRecv; /* 속성 localRecv */ };

MqttMgr 개체의 데이터는 접근자 또는 돌연변이자 작업을 사용하지 않고 직접 액세스 및 수정됩니다. 예를 들어, 클라이언트와 localRecv는 MqttMgr의 인스턴스를 가리키는 me 포인터를 통해 액세스됩니다.

mqtt_recvMsgError(&me->클라이언트, &me->localRecv);

MqttMgr에는 표 1에 표시된 속성 목록이 있습니다.

표 1. MqttMgr 속성

그림 2의 구조는 관련 행위자 간의 관계를 염두에 두는 데 유용합니다. 브로커에게 정보를 보내려는 Collector 객체, 네트워크를 다루는 NetMgr 개체, 및 MqttMgr 개체.

그림 2. IoT 시스템 구조 초안(출처:VortexMakes)

그림 3의 시퀀스 다이어그램은 MQTT 서버와의 세션을 열어야 할 때 MqttMgr 객체가 나머지 시스템과 상호 작용하는 방식을 보여줍니다. 이 다이어그램에서 MqttMgr 상태와 교환된 비동기 메시지는 Collector, MqttMgr 및 NetMgr 액터 간에 표시됩니다.

그림 3. MQTT 브로커에 연결(출처:VortexMakes)

NetMgr 개체가 브로커에 대해 네트워크 연결을 설정한 후 MqttMgr에서 MQTT 서버로 보낸 첫 번째 패킷은 CONNECT 패킷이어야 합니다. 따라서 Collector 액터는 Connect(clientId, keepAlive) 이벤트를 MqttMgr 액터에 보냅니다. 이 이벤트는 클라이언트 식별자와 연결 유지 시간 간격을 전달해야 합니다. 서버가 연결 요청을 수락하면 MqttMgr 액터는 ConnAccepted 이벤트를 Collector 액터에게 보내 이 상황을 알립니다. 그때부터 Collector 액터는 해당 브로커에 정보 메시지를 게시할 수 있습니다.

서버가 연결 요청을 거부하면 MqttMgr 액터는 ConnRefused 이벤트를 Collector 액터에게 보냅니다. 이 이벤트에는 그림 4와 같이 거부 원인을 알리는 코드가 포함됩니다. MQTT v3.1.1 섹션 3.2.2.3을 참조하세요.

그림 4. 브로커가 연결 요청을 거부함(출처:VortexMakes)

그림 5는 메시지가 게시될 때의 상호 작용 흐름을 보여줍니다. 이를 위해 Collector 액터는 Publish(data, size, topic, qos) 이벤트를 전송합니다. 이 이벤트는 게시할 정보(data), 정보의 길이(바이트), topic 이름을 전달합니다. 정보(주제)와 이 메시지를 전달하기 위한 보증 수준(qos)이 게시됩니다. 앞서 언급한 IoT 장치에서 게시된 정보는 JSON 사양을 사용하여 형식이 지정되었습니다. 사람이 읽을 수 있는 텍스트의 속성-값 쌍이 있는 데이터 개체를 포함하는 개방형 표준 형식입니다. 이 형식은 C로 작성된 간단하고 가벼운 라이브러리인 jWrite를 사용하여 수행되었습니다.

그림 5. 브로커에 데이터 게시(출처:VortexMakes)

그림 6은 네트워크로의 MQTT 메시지 수신 및 전송이 실패하는 시나리오를 보여줍니다. 네트워크 관리자가 네트워크에서 트래픽을 수신할 수 없으면 MqttMgr 액터에 ReceiveFail을 보냅니다. 마찬가지로 네트워크 관리자가 네트워크에 데이터를 보낼 수 없으면 MqttMgr 액터에 SendFail을 보냅니다.

그림 6. 네트워크 장애(출처:VortexMakes)

표 2는 표시된 시나리오에서 관련된 이벤트를 요약한 것입니다.

표 2. 이벤트

결론

이 기사에서 제안한 MQTT-Reactive 라이브러리, 상태 머신 및 소프트웨어 아키텍처는 세마포어, 뮤텍스, 지연 또는 이벤트 플래그와 같은 기존 차단 메커니즘의 위험을 피함으로써 반응형 임베디드 시스템이 새로운 MQTT 클라이언트를 구현할 수 있도록 합니다. 방법. 이는 활성 개체라는 동시성 단위 내에서 MQTT-Reactive 코드를 캡슐화하여 달성됩니다. , 상태 기반 동작이 제안된 상태 시스템에 정의되어 있습니다. 이 활성 개체는 네트워크를 통한 트래픽 수신 및 전송뿐만 아니라 사물 인터넷 애플리케이션용 서버에 연결 및 정보 게시에 사용되는 비동기 이벤트를 교환하여 시스템의 나머지 부분과 통신합니다.