今回の目標

Eclipse Pahoを用いてAWS IoT CoreとMQTT通信するプログラムをC言語で実装することを目指す。
まずは同期でAWS IoT Coreにメッセージをpublishするプログラムの作成を行う(Payloadも単純な文字列)。

非同期通信、メッセージのsubscribe、json形式のメッセージ作成はまた別の機会に。

実施環境

Virtualbox上にUbuntuを構築して実施
- VirtualBox 6.1.30
- Ubuntu 20.04.3 LTS
AWS IoT Coreは2021/11時点。

(最終的にはRaspberry Piなどのデバイスを用いる予定)

事前準備

pahoのインストール

最初にmqttクライアントライブラリであるpahoをインストールする。

sudo apt install build-essential gcc make cmake cmake-gui cmake-curses-gui
sudo apt install fakeroot fakeroot devscripts dh-make lsb-release
sudo apt install libssl-dev

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

インストールが成功していれば、/usr/local/lib以下にlibmqtt*.soが、/usr/local/include以下にMQTT*.hが存在する。

AWS IoT Coreへの「モノ」の登録

AWS IoT Core内部の「管理」→「モノ」を選択する。
「モノ」のページ内部から「モノの作成」を選択し、通信先となるモノを作成する。今後も見据え、名前付きシャドウも作成した。
モノと同時に作成したポリシーは以下の通り。

{
  "Version": "YYYY-MM-XX",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:XXX:client/myDevice"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "arn:aws:iot:XXX:topic/myIoT/sensor"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:XXX:topicfilter/myIoT/cmd"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Receive",
      "Resource": "arn:aws:iot:XXX:topic/myIoT/cmd"
    },
    {
      "Effect": "Allow",
      "Action": "iot:UpdateThingShadow",
      "Resource": "arn:aws:iot:XXX:thing/myDeviceShadow"
    },
    {
      "Effect": "Allow",
      "Action": "iot:GetThingShadow",
      "Resource": "arn:aws:iot:XXX:thing/myDeviceShadow"
    }
  ]
}

モノの作成後、ルート証明書など一式をダウンロードしておく。
あとで作成する通信プログラム内でpahoからこれらのファイルを読み取って通信する。

通信プログラムの作成

pahoを用いたプログラムのソースコード。
Paho MQTT C Client Libraryの"Synchronous publication example"をベースに作成した。

pub_sync.c

/* pub_sync.c */
# include <stdio.h>
# include <stdlib.h>
# include <string.h>
# include "MQTTClient.h"
# include "mqtt_var.h"

int main(int argc, char* argv[]){
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = conn_opts_initial;
    MQTTClient_SSLOptions ssl_opts = ssl_opts_initial;
    MQTTClient_message pubmsg = {
         /* strcture detail is described on "MQTTClient.h" */
        {'M', 'Q', 'T', 'M'}, // the eyecather(must be MQTM)
        1,      // version number of the structure
        0,      // payload lenghth in bytes
        NULL,   // payload of MQTT message (void*)
    };
    MQTTClient_deliveryToken token;
    int rc;

    /* MQTTCLIENT_PERSISTENCE_NONE -> memory-based persistence mechanism */
    if((rc = MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS){
        printf("Failed to create client, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    conn_opts.ssl = &ssl_opts;

    if((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS){
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    char* payload = "This is MQTT message";
    pubmsg.payload = (void*)payload;
    pubmsg.payloadlen = (int)strlen(payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    if((rc = MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token)) != MQTTCLIENT_SUCCESS){
        printf("Failed to publsh message, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    printf("Wait for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT / 1000), payload, PUB_TOPIC, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);

    if((rc = MQTTClient_disconnect(client, 10000)) != MQTTCLIENT_SUCCESS){
        printf("Failed to disconnect, return code %d\n", rc);
    }
    MQTTClient_destroy(&client);
    return rc;
}

aws_cert.h

/* aws_cert.h */
# ifndef __AWS_CERT_H__
# define __AWS_CERT_H__

# define ADDRESS "ssl://XXX.amazonaws.com:8883"


# define CERFILE_PATH "<AWSからダウンロードした証明書ファイルディレクトリへのパス>"

# define root_ca CERFILE_PATH "AmazonRootCA1.pem"
# define certificate  CERFILE_PATH "XXX-certificate.pem.crt"
# define private_key  CERFILE_PATH "XXX-private.pem.key"
# define ca_path CERFILE_PATH "ca/"

# endif

mqtt_var.h

/* mqtt_var.h */
# ifndef __MQTT_VAR__
# define __MQTT_VAR__

# include "MQTTClient.h"
# include "aws_cert.h"

const char* CLIENTID = "myDevice";
const char* SUB_TOPIC = "myIoT/cmd";
const char* PUB_TOPIC = "myIoT/sensor";

const int QOS = 1;
const long TIMEOUT = 10000L;

MQTTClient_connectOptions conn_opts_initial = {
    /* strcture detail is described on "MQTTClient.h" */
    {'M', 'Q', 'T', 'C'}, // the eyecather(must be MQTC)
    8,      // version of structure
    60,     // keep alive interval
    1,      // clean session
    1,      // reliable
    NULL,   // MQTTClient_willOptions
    NULL,   // username
    NULL,   // password
    30,     // connection timeout
    0,      // retry Interval
    NULL,   // MQTTClient_SSLOptions
    0,      // the number of serverURI 
    NULL,   // serverURIs
    MQTTVERSION_DEFAULT, // MQTT version
    {NULL, 0, 0}, // Returned from the connect when the MQTT version used to connect is 3.1.1
    {0, NULL}, // binary password
    -1, // maximum number of messages in flight
    0,      // MQTT c5 clean start flag
    NULL,   // HTTP headers for websockets
    NULL,   // HTTP proxy for websockets
    NULL,   // HTTPs proxy for websockets
};

MQTTClient_SSLOptions ssl_opts_initial = {
    {'M', 'Q', 'T', 'S'}, // the eyecatcher(must be MQTS)
    3,      // struct version
    root_ca,   // trustStore (PEM format containig public digital certificates)
    certificate,   // keyStore (PEM format containing public certificate chain)
    private_key,   // private key
    NULL,   // password of private key
    NULL,   // the list of ciper suites
    1,      // true/false option to enbale verifaication of server certificate
    MQTT_SSL_VERSION_TLS_1_2, // SSL/TLS version to use
    1,      // whether to carry out post-connect checks
    ca_path,   // CA path
    NULL,   // callback function for OpenSSL error handler
    NULL,   // application-specific contex for OpenSSL error handler
    NULL,   // callback function for setting TLS-PSK options
    NULL,   //Application-specific contex for ssl_psk_cb
    0,      // disable default trust store
    NULL,   // protocol-list
    0,      // protocal length of protocol-list vector
};

# endif

ビルドコマンドは以下。libpaho-mqtt*.soのリンク設定を忘れずに行う。

gcc pub_sync.c -lpaho-mqtt3cs -o pub_sync

動作確認

AWS IoT Coreでは「MQTT テストクライアント」を用いることでメッセージ通信を確認することができる。

「トピックをサブスクライブする」からトピックフィルターを設定
- フィルターとして「myIoT/#」を設定

正常に送信できれば、AWS IoT Core上で”This is MQTT message”がPayloadのメッセージ受信を確認することができる。

このときのクライアント側の出力は次のようになる。

$ ./pub_sync
Wait for up to 10 seconds for publication of This is MQTT message
on topic myIoT/sensor for client with ClientID: myDevice
Message with delivery token 1 delivered

次回にやること

今回はPayloadとして文字列をそのまま送信したため、AWS IoT Core上でエラーメッセージが発生する。
次回はこれを解決するために送信するデータをJSON形式に変換してAWS IoT Coreにデータ送信する。

参考