Paho C を使ったデプロイメントへの接続

本記事では、Cプロジェクトで Eclipse Paho C を使用し、クライアントと MQTT ブローカー間の接続、サブスクライブ、メッセージ送受信、サブスクリプション解除などの機能を実装する方法を主に紹介します。

Eclipse Paho Embedded C はデスクトップOSでも使用可能ですが、主に mbed、Arduino、FreeRTOS などの組み込み環境向けです。

前提条件

Paho C クライアントを使用してアプリケーションを EMQX Cloud に接続する前に、MQTT ブローカーをデプロイしていることを確認してください。

MQTT ブローカーのデプロイ

アプリケーションをEMQX Cloudに接続するには、デプロイメントを作成して設定する必要があります。

サーバレスデプロイメント

EMQX Cloudコンソールでサーバレスデプロイメントを作成します。
デプロイメントが作成されて稼働したら、デプロイメントの概要ページに移動し、以下を含むMQTT接続情報を確認します。
- ブローカーアドレス
- ポート番号（サーバレスではTLSポートのみ対応）
サーバレスデプロイメントはTLS接続が必須です。概要ページからCA証明書をダウンロードし、TLS用のポート8883を使用してください。
デプロイメントのアクセス制御 -> クライアント認証で、デフォルト認証（ユーザー名／パスワード）を設定します。

詳細はサーバレスポートガイドを参照してください。

Dedicated Flex または BYOC デプロイメント

EMQX CloudコンソールでDedicated FlexまたはBYOCのデプロイメントを作成できます。
作成後、デプロイメントの概要ページに移動し、以下を含むMQTT接続情報を取得します。
- ブローカーアドレス
- MQTTおよびWebSocket用のTCPおよびTLSポート番号（TCPおよびTLS接続の両方に対応）
デプロイメントのアクセス制御 -> クライアント認証で、デフォルト認証（ユーザー名／パスワード）を設定します。

詳細なポート設定については、Dedicated & BYOCポートガイドをご覧ください。

インストール依存関係

bash

sudo apt-get update
sudo apt-get -y install build-essential git cmake

ソースからのビルド

継続的インテグレーションビルドは、Linux と Mac 用に Travis-CI、Windows 用に AppVeyor で提供されています。

Linux/Mac

bash

git clone https://github.com/eclipse/paho.mqtt.c.git
cd org.eclipse.paho.mqtt.c.git
make
sudo make install

Windows

bash

mkdir build.paho

cd build.paho

call "C:\Program Files (x86)\Microsoft Visual Studio 14.0\VC\vcvarsall.bat" x64

cmake -G "NMake Makefiles" -DPAHO_WITH_SSL=TRUE -DPAHO_BUILD_DOCUMENTATION=FALSE -DPAHO_BUILD_SAMPLES=TRUE -DCMAKE_BUILD_TYPE=Release -DCMAKE_VERBOSE_MAKEFILE=TRUE ..

nmake

TCP プロトコルでの接続

このセクションでは、TCP ポートを介して MQTT ブローカーに接続する方法を紹介します。

依存ライブラリのインクルード

#include "stdlib.h"
#include "string.h"
#include "unistd.h"
#include "MQTTClient.h"

接続先アドレス、認証情報、メッセージのパブリッシュおよび受信トピックの定義

// SSL/TLS 接続の有効化（1：SSL/TLS、0：TCP）
#define USE_SSL 0

#if USE_SSL
#define ADDRESS "ssl://broker.emqx.io:8883"
#else
#define ADDRESS "tcp://broker.emqx.io:1883"
#endif

#define USERNAME    "emqx"
#define PASSWORD    "public"
#define CLIENTID    "c-client"
#define QOS         0
#define TOPIC       "emqx/c-test"
#define TIMEOUT     10000L

ブローカーへの接続

int rc;
MQTTClient client;

MQTTClient_create(&client, ADDRESS, CLIENTID, 0, NULL);
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
conn_opts.username = USERNAME;
conn_opts.password = PASSWORD;

if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
    printf("接続に失敗しました。戻り値コード %d\n", rc);
    exit(-1);
} else {
    printf("MQTT ブローカーに接続しました！\n");
}

SSL/TLS プロトコルでの接続

このセクションでは、SSL/TLS ポートを介して MQTT ブローカーに接続する方法を紹介します。

接続先アドレスを定義し、USE_SSL を 1 に設定します。

// SSL/TLS 接続の有効化（1：SSL/TLS、0：TCP）
#define USE_SSL 1

SSL/TLS によるブローカーへの接続

configureSSLOptions() 関数で SSL オプションを設定します。サーバー証明書を設定する場合は、ssl_opts.trustStore に証明書をロードしてください。不要な場合は ssl_opts.enableServerCertAuth を 0 に設定します。

MQTTClient_SSLOptions configureSSLOptions() {
    MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
    ssl_opts.enableServerCertAuth = 1;
    // CA_CERTIFICATE_FILE_Path にサーバー CA を指定
    // ssl_opts.trustStore = CA_CERTIFICATE_FILE_Path;
    return ssl_opts;
}

···

#if USE_SSL
 MQTTClient_SSLOptions ssl_opts = configureSSLOptions();
 conn_opts.ssl = &ssl_opts;
#endif

パブリッシュとサブスクライブ

このセクションでは、MQTT ブローカーへの接続に成功した後のトピックのサブスクライブとメッセージのパブリッシュ方法を紹介します。

トピックのサブスクライブ

サブスクライブするトピックとトピックの QoSレベルを設定します。

// トピックのサブスクライブ
MQTTClient_subscribe(client, TOPIC, QOS);

トピックのサブスクリプション解除

以下のコードでトピックのサブスクリプションを解除します。解除するトピックを定義する必要があります。

MQTTClient_unsubscribe(client, TOPIC);

メッセージのパブリッシュ

メッセージをパブリッシュする際に、MQTT ブローカーにトピックとペイロードを通知します。

void publish(MQTTClient client, char *topic, char *payload) {
    MQTTClient_message message = MQTTClient_message_initializer;
    message.payload = payload;
    message.payloadlen = strlen(payload);
    message.qos = QOS;
    message.retained = 0;
    MQTTClient_deliveryToken token;
    MQTTClient_publishMessage(client, topic, &message, &token);
    MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("トピック `%s` に `%s` を送信しました\n", TOPIC, payload);
}

// メッセージのパブリッシュを完了するために `publish` 関数をループで呼び出す
char payload[16];
for (int i = 0; i < 100; i += 1) {
    // ブローカーにメッセージをパブリッシュ
    snprintf(payload, 16, "message-%d", i);
    publish(client, TOPIC, payload);
    sleep(1);
}

メッセージの受信

以下のコードは、クライアントがメッセージイベントを監視し、メッセージを受信した後にコールバック関数を実行して、受信したメッセージとそのトピックをコンソールに出力することを指定しています。

MQTTClient_setCallbacks(client, NULL, NULL, on_message, NULL);
···
// サブスクライブしたトピックで受信したメッセージの内容を表示する on_message コールバック関数の定義
int on_message(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    char *payload = message->payload;
    printf("トピック `%s` から `%s` を受信しました\n", topicName, payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

MQTT ブローカーからの切断

クライアントが能動的に切断したい場合は、以下のコードを使用します。

MQTTClient_disconnect(client, TIMEOUT);
MQTTClient_destroy(&client);

上記は主要なコードのみを抜粋しています。プロジェクトの完全なコードは mqtt-client-C-paho にて公開されていますので、ダウンロードしてご利用ください。

完全なコード例

#include "stdlib.h"
#include "string.h"
#include "unistd.h"
#include "MQTTClient.h"

// SSL/TLS 接続の有効化（1：SSL/TLS、0：TCP）
#define USE_SSL 1

#if USE_SSL
#define ADDRESS "ssl://broker.emqx.io:8883"
#else
#define ADDRESS "tcp://broker.emqx.io:1883"
#endif

#define USERNAME    "emqx"
#define PASSWORD    "public"
#define CLIENTID    "c-client"
#define QOS         0
#define TOPIC       "emqx/c-test"
#define TIMEOUT     10000L

MQTTClient_SSLOptions configureSSLOptions() {
    MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
    ssl_opts.enableServerCertAuth = 1;
    ssl_opts.trustStore = CA_CERTIFICATE_FILE;
    return ssl_opts;
}


void publish(MQTTClient client, char *topic, char *payload) {
    MQTTClient_message message = MQTTClient_message_initializer;
    message.payload = payload;
    message.payloadlen = strlen(payload);
    message.qos = QOS;
    message.retained = 0;
    MQTTClient_deliveryToken token;
    MQTTClient_publishMessage(client, topic, &message, &token);
    MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("トピック `%s` に `%s` を送信しました\n", TOPIC, payload);
}

int on_message(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    char *payload = message->payload;
    printf("トピック `%s` から `%s` を受信しました\n", topicName, payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

int main(int argc, char *argv[]) {
    int rc;
    MQTTClient client;

    MQTTClient_create(&client, ADDRESS, CLIENTID, 0, NULL);
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.username = USERNAME;
    conn_opts.password = PASSWORD;

#if USE_SSL
    MQTTClient_SSLOptions ssl_opts = configureSSLOptions();
    conn_opts.ssl = &ssl_opts;
#endif

    MQTTClient_setCallbacks(client, NULL, NULL, on_message, NULL);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("接続に失敗しました。戻り値コード %d\n", rc);
        exit(-1);
    } else {
        printf("MQTT ブローカーに接続しました！\n");
    }
    // トピックをサブスクライブ
    MQTTClient_subscribe(client, TOPIC, QOS);
    
    char payload[16];
    for (int i = 0; i < 10; i += 1) {
        // ブローカーにメッセージをパブリッシュ
        snprintf(payload, 16, "message-%d", i);
        publish(client, TOPIC, payload);
        sleep(1);
    }
    MQTTClient_unsubscribe(client, TOPIC);
    MQTTClient_disconnect(client, TIMEOUT);
    MQTTClient_destroy(&client);
    return rc;
}

テスト

CMakeLists.txt ファイルを作成します。

cmake_minimum_required(VERSION 3.17)
find_package(eclipse-paho-mqtt-c 1.3.9 REQUIRED)
project(mqtt_c C)
include_directories(/usr/local/include)
link_directories(/usr/local/lib)
set(CMAKE_C_STANDARD 99)
add_executable(mqtt_c main.c)
target_link_libraries(mqtt_c paho-mqtt3c)

コードをコンパイルして実行します。

まとめ

本記事では、Cプロジェクトで MQTT 接続の作成、サブスクライブ、メッセージの送受信、サブスクリプション解除、切断を実装する方法を紹介しました。サンプルのソースコードはこちらからダウンロード可能です。また、他言語のデモ例も GitHub にて多数公開されています。

専用フレックス

VPC ピアリング接続

PrivateLink の設定

拡張認証

拡張認可

ルール

LLM ベースの MQTT データ処理

サーバレスデプロイメント API

Paho C を使ったデプロイメントへの接続 ​

前提条件 ​

MQTT ブローカーのデプロイ ​

サーバレスデプロイメント ​

Dedicated Flex または BYOC デプロイメント ​

インストール依存関係 ​

ソースからのビルド ​

Linux/Mac ​

Windows ​

TCP プロトコルでの接続 ​

SSL/TLS プロトコルでの接続 ​

パブリッシュとサブスクライブ ​

トピックのサブスクライブ ​

トピックのサブスクリプション解除 ​

メッセージのパブリッシュ ​

メッセージの受信 ​

MQTT ブローカーからの切断 ​

完全なコード例 ​

テスト ​

まとめ ​

Paho C を使ったデプロイメントへの接続

前提条件

MQTT ブローカーのデプロイ

サーバレスデプロイメント

Dedicated Flex または BYOC デプロイメント

インストール依存関係

ソースからのビルド

Linux/Mac

Windows

TCP プロトコルでの接続

SSL/TLS プロトコルでの接続

パブリッシュとサブスクライブ

トピックのサブスクライブ

トピックのサブスクリプション解除

メッセージのパブリッシュ

メッセージの受信

MQTT ブローカーからの切断

完全なコード例

テスト

まとめ