coreMQTT 接続共有デモ - FreeRTOS
序章ソースコード機能Typedefsメインタスク同期パブリッシャータスク非同期パブリッシャータスクサブスクライバータスク

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

coreMQTT 接続共有デモ

重要

これは、FreeRTOS リリース 202012.00 で使用する FreeRTOS ユーザーガイドのアーカイブ版です。このドキュメントの最新バージョンについては、FreeRTOS ユーザーガイドを参照してください。

序章

coreMQTT 接続共有デモプロジェクトでは、マルチスレッドアプリケーションを使用して、クライアントとサーバー間の相互認証で TLS を使用して AWS MQTT ブローカーへの接続を確立する方法を示します。このデモでは、mbedTLS ベースのトランスポートインターフェイスの実装を使用して、サーバーとクライアントの認証 TLS 接続を確立し、QoS 1 レベルでの MQTT のサブスクライブおよび公開ワークフローについて説明します。デモでは、トピックフィルターをサブスクライブし、フィルターに一致するトピックに公開し、QoS 1 レベルでサーバからそれらのメッセージを受信するまで待機します。ブローカーに公開し、ブローカーから同じメッセージを受信するこのサイクルは、無限に繰り返されます。このデモのメッセージは QoS 1 で送信され、MQTT 仕様に従って少なくとも 1 つの配信が保証されます。

注記

FreeRTOS デモをセットアップして実行するには、FreeRTOS の開始方法 の手順に従います。

このデモでは、スレッドセーフなキューを使用して MQTT API を操作するためのコマンドを保持します。このデモでは、注意すべきタスクが 4 つあります。

  • コマンド (メイン) タスクはコマンドキューからコマンドを受け取り、それらを処理します。他のタスクでは、処理対象のコマンドをこのキューに配置します。このタスクはループに入り、その間にコマンドを処理します。終了コマンドを受信すると、このタスクはループから抜け出します。

  • 同期パブリッシャータスクは、一連の公開オペレーションを作成し、コマンドキューにプッシュします。これらのオペレーションは、コマンドタスクによって実行されます。このタスクでは同期パブリッシングを使用します。つまり、このタスクは各公開オペレーションが完了するまで待機してから、次の公開をスケジュールします。

  • 非同期パブリッシャータスクは、一連の公開オペレーションを作成し、コマンドキューにプッシュします。これらのオペレーションは、コマンドタスクによって実行されます。このタスクは前のタスクと異なり、公開オペレーションが完了するのを待つことなく次のタスクをスケジュールします。すべての公開オペレーションがキューに追加されてから、各公開オペレーションのステータスをチェックします。同期パブリッシングと非同期パブリッシングの違いは、これらのタスクの動作のみであることに注意してください。実際の公開コマンドに違いはありません。

  • サブスクライバータスクは、2 つのパブリッシャータスクが発行するすべてのメッセージのトピックに一致するトピックフィルターへの MQTT サブスクリプションを作成します。このタスクはループに入り、他のタスクによって発行されたメッセージの受信を待ちます。

タスクは、受信したメッセージを保持するキューを持つことができます。コマンドタスクは、受信トピックにサブスクライブされているタスクのキューに受信メッセージをプッシュします。

このデモでは、相互認証を使用した TLS 接続を使用して AWSに接続します。デモ中にネットワークが予期せず切断された場合、クライアントはエクスポネンシャルバックオフロジックを使用して再接続を試みます。クライアントが正常に再接続してもブローカーが前回のセッションを再開できない場合、クライアントは前回のセッションと同じトピックに再度サブスクライブします。

シングルスレッドとマルチスレッド

coreMQTT の使用モデルは、シングルスレッドとマルチスレッド (マルチタスク) の 2 つがあります。このデモでは、独自のマルチスレッドスキームを作成する方法を説明します。エージェント (またはデーモン) タスク内で MQTT プロトコルをバックグラウンドで実行するマルチスレッドの例もあります。詳細については、MWTT エージェントと coreMQTT を使用したデモを参照してください。エージェントタスクで MQTT プロトコルを実行する場合、MQTT のステータスを明示的に管理する必要や、MQTT_ProcessLoop 関数を呼び出す必要はありません。また、エージェントタスクを使用する場合、ミューテックスなどの同期プリミティブを必要とせずに、複数のアプリケーションタスクが単一の MQTT 接続を共有できます。

ソースコード

デモのソースファイルの名前は mqtt_demo_connection_sharing.c で、freertos/demos/coreMQTT/ ディレクトリと GitHub ウェブサイトで手に入ります。

機能

このデモでは、合計 4 つのタスクを作成します。そのうち 3 つは MQTT API コールを要求するもの、残り 1 つはそれらのリクエストを処理するプライマリタスクです。このデモでは、3 つのサブタスクを作成し、処理ループを呼び出し、その後クリーンアップするループにプライマリタスクが入ります。プライマリタスクは、サブタスク間で共有されるブローカーへの MQTT 接続を 1 つ作成します。2 つのサブタスクはメッセージをブローカーに発行し、3 つ目のサブタスクは発行されたメッセージのすべてのトピックに一致するトピックフィルターへの MQTT サブスクリプションを使用してメッセージを受信します。

Typedefs

デモでは、次の構造、列挙型、関数ポインタを定義しています。

コマンド

タスクは MQTT API コールを直接行うのではなく、Command_t 構造を使って、メインタスクに適切な API オペレーションを呼び出すように指示するコマンドを作成します。コマンドには次のタイプがあります。

  • PROCESSLOOP

  • PUBLISH

  • SUBSCRIBE

  • UNSUBSCRIBE

  • PING

  • DISCONNECT

  • RECONNECT

  • TERMINATE

TERMINATE コマンドには対応する MQTT API オペレーションがありません。デモでは、メインタスクにコマンドの処理を停止し、クリーンアップオペレーションを開始するように指示するために使用されます。例えば、公開やサブスクライブ情報などの追加情報が一部の MQTT コマンド (MQTT_Publish、MQTT_Subscribe、MQTT_Unsubscribe など) に必要になるため CommandContext_t フィールドを使用します。このフィールドはこれら 3 つのコマンドでは必須で、その他のコマンドではオプションです。

これらのコマンドにはこのコンテキストが必要であるため、コマンドがキューに配置された後は、コマンドが完了するまでこの値を変更しないでください。コマンドが完了すると、オプションのコールバックを呼び出すことができます。このデモでは、タスク通知を作成するコールバックを使用して、コマンドが完了したことを呼び出し側のタスクに通知します。確認を必要とする MQTT オペレーション (QoS が 0 より大きいサブスクライブ、サブスクライブ解除、公開) の場合、確認が受信されるとコマンドが完了したと見なされます。それ以外の場合は、対応する MQTT API コールが返された時点でコマンド完了です。

mqtt_demo_connection_sharing.c ファイルに以下の定義があります。

確認

一部の MQTT オペレーションは確認を必要とするため、AckInfo_t の配列を使用します。これには、完了コールバックを呼び出せるように、期待される確認のパケット識別子、それを待機している元のコマンドが含まれます。

サブスクリプション

このデモでは、各タスクのサブスクリプションを追跡できます。そのためには、サブスクリプションを要求する各タスクは、発行されたメッセージを受信するメッセージキューを提供する必要があります (SubscriptionElement_t)。複数のタスクが同じトピックフィルターにサブスクライブできます。これは、別々のレスポンスキューを使用することが想定されるためです。

発行済みメッセージの受信

タスクはメインタスクと並行して実行されるため、サブスクライブされた各タスクが受信した公開メッセージを読み取るのを待たなければならないのは、メインタスクにとって難しく時間がかかります。したがって、受信した各メッセージは、発行されたメッセージのトピック (PublishElement_t) にサブスクライブされているタスクのレスポンスキューにコピーされます。MQTT クライアントから受信した公開パケットにはクライアントのネットワークバッファへのポインタが含まれているため、受信メッセージのペイロードとトピック名は応答キューに挿入される前に別のバッファにコピーされます。この方法では、MQTT クライアントがネットワークバッファを消去した後も、サブスクライブされたタスクは受信した情報を読み取ることができます。

メインタスク

メインアプリケーションタスク RunCoreMQTTConnectionSharingDemo は、永続的な MQTT セッションを確立し、3 つのサブタスクを作成し、終了コマンドを受信するまで処理ループを実行します。永続セッションが使用されるため、ネットワークが予期せず切断された場合、デモはブローカーからのサブスクリプションや受信した公開メッセージを失うことなく、バックグラウンドでブローカーに再接続します。実行のたびに新しい永続セッションを作成するには、デモはブローカに clean session フラグを設定して接続し、フラグを未設定にして切断して再接続します。処理ループが終了するとブローカから切断され、ネットワークの再接続が行われた時点から再びループします。

デモが正常に完了すると、次の図のような出力が生成されます。

正常に完了した場合のデモ端末の出力を共有する MQTT 接続
コマンドループ

コマンドループ prvCommandLoop は、コマンドがコマンドキューに配置されるまで待機してから、適切な MQTT API を呼び出します。DISCONNECT と TERMINATE 以外のすべてのコマンドで、MQTT_ProcessLoop も呼び出されます。このデモでは、ソケットウェイクアップコールバックを設定し、データがソケットで利用可能になったときに PROCESSLOOP コマンドをキューに送信します。ただし、その時点でキュー内に先行するコマンドが多数存在する可能性があります。他のコマンドが処理されている間に受信データを無視しないようにするには、各コマンドの後に MQTT_ProcessLoop が 1 回のイテレーションに対して呼び出されます。

コマンドの処理

詳しくは、prvProcessCommand 関数を参照してください。

同期パブリッシャータスク

同期パブリッシャータスク prvSyncPublishTask は PUBLISH オペレーションを同期的に作成し、各オペレーションが完了するのを待ってから次のオペレーションをスケジュールします。このデモでは QoS 1 を使用してメッセージを公開します。つまり、公開確認パケットが受信されるまで、これらのオペレーションは完了したと見なされません。

非同期パブリッシャータスク

非同期パブリッシャータスク prvAsyncPublishTask は、公開の完了を待たずに次の公開をキューに配置します。つまり、必ずしも MQTT オペレーションが完了するのを待ってからタスクを再開しなければならないとは限りません。各公開コマンドには独自のコンテキスト構造体が必要なため、このタスクは同期パブリッシャータスクのように単一のコンテキスト構造を再利用することはできません。なぜなら、前のコマンドでもそのコンテキスト構造が必要になる可能性があるためです。したがって、各コンテキスト構造にメモリを割り当てて、発行されるすべてのメッセージがキューに配置された後に、割り当てられたすべてのメモリを解放するのを待機します。

サブスクライバータスク

サブスクライバータスク、prvSubscribeTask は、同期タスクおよび非同期タスクから発行されたメッセージのすべてのトピックに一致するトピックフィルターをサブスクライブします。その後、発行されたすべてのメッセージを受信してから、サブスクライブを解除します。このタスクは、コマンドループの終了をメインタスクに通知する TERMINATE オペレーションの作成も担当します。