Anypoint Studio を使用した MQTT Connector 1.0 の設定

Anypoint Studio (Studio) エディターは、Mule アプリケーション、プロパティ、および設定ファイルの設計と更新に役立ちます。

Studio でコネクタを追加および設定する手順は、次のとおりです。

Mule プロジェクトを作成します。
コネクタを Mule プロジェクトに追加します。
コネクタのフローのソースを設定します。
フローにコネクタの操作を追加します。
コネクタのグローバル要素を設定します。
接続プロトコルを設定します。
認証用のログイン情報を設定します。
フェールオーバーサーバーリストを設定します。
ファイル永続性を有効化します。
接続の詳細オプションを設定します。

「アプリケーションログの表示」で説明されているように、コネクタを実行するときに、アプリケーションログを表示して問題を確認できます。

Studio で初めてコネクタを設定する場合は、「Anypoint Studio を使用したコネクタの設定」を参照してください。このトピックを参照した後、コネクタ項目についての詳細情報が必要な場合は、「MQTT Connector リファレンス」を参照してください。

Mule プロジェクトを作成する

Studio で、コネクタを追加および設定する新しい Mule プロジェクトを作成します。

Studio で、[File (ファイル)] > [New (新規)] > [Mule Project (Mule プロジェクト)] を選択します。
Mule プロジェクトの名前を入力して、[Finish (完了)] をクリックします。

コネクタを Mule プロジェクトに追加する

MQTT 用 Anypoint Connector (MQTT Connector) を Mule プロジェクトに追加して、XML コードにコネクタの名前空間およびスキーマの場所を自動的に入力し、プロジェクトの pom.xml ファイルに必須の連動関係を追加します。

[Mule Palette (Mule パレット)] ビューで、[(X) Search in Exchange ((X) Exchange 内を検索)] をクリックします。
[Add Dependencies to Project (連動関係をプロジェクトに追加)] ウィンドウで、検索項目に「mqtt」と入力します。
[Available modules (使用可能なモジュール)] で [MQTT 3 Connector (MQTT 3 Connector)] をクリックします。
[Add (追加)] をクリックします。
[Finish (完了)] をクリックします。

Studio でコネクタを Mule プロジェクトに追加しても、Studio ワークスペースの他のプロジェクトはそのコネクタを使用できません。

ソースの設定

ソースは、指定された条件が満たされたときにフローを開始します。次の入力元のいずれかを設定して、MQTT Connector で使用できます。

[MQTT3] > [On New Message]
受信メッセージをリスンすることでフローを開始する
[HTTP] > [Listener]
設定されたホストとポートで要求を受信するたびにフローを開始する
Scheduler
時間ベースの条件が満たされたときにフローを開始する

たとえば、On New Message ソースを設定する手順は、次のとおりです。

[Mule Palette (Mule パレット)] ビューで、[MQTT3] > [On New Message] を選択します。
[On New Message] を Studio キャンバスにドラッグします。
[On New Message] 設定画面で、必要に応じて [Display Name (表示名)] 項目の値を変更します。
[Connector configuration (コネクタ設定)] 項目の横にあるプラス記号 (+) をクリックして、アプリケーション内のすべてのソースのインスタンスで使用できるグローバル要素を設定します。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で、この設定に指定する接続種別のいずれかを選択します。
- MQTT3 URL Connection (MQTT3 URL 接続)
- MQTT3 Fail-Over Connection (MQTT3 フェールオーバー接続)
- MQTT3 Form Connection (MQTT3 フォーム接続)

ブローカーへの MQTT 接続を識別する一意の直観的な [Client ID (クライアント ID)] を指定します。
[General (一般)] タブで接続情報 (ブローカーで必要なライブラリなど) も指定します。
[LWT] タブで、必要に応じて LWT (Last Will and Testament) メッセージを指定します。
[TLS/SSL] タブで、必要に応じて TLS 設定を指定します。
[Advanced (詳細)] タブで、必要に応じて接続の詳細オプションと再接続戦略を指定します。
[OK] をクリックします。
[On New Message] 設定画面の [Topics (トピック)] で [Edit inline (インライン編集)] を選択し、リスナーでサブスクライブするトピックをリストします。
プラス記号 (+) をクリックして、トピックを設定します。
[Topic (トピック)] 設定画面で、トピックの単一レベルまたはマルチレベルのサブスクリプションを表す [Topic filter (トピック検索条件)] を設定します。
[QoS] を次のいずれかのサービス品質レベルに設定します。
- AT_MOST_ONCE
- AT_LEAST_ONCE
- EXACTLY_ONCE
[Finish (完了)] をクリックします。

フローにコネクタの操作を追加する

フローにコネクタの操作を追加すると、そのコネクタで実行する特定の操作をすぐに定義できます。

MQTT Connector の操作を追加するには、次の手順に従います。

[Mule Palette (Mule パレット)] ビューで [MQTT3] を選択し、目的の操作を選択します。
その操作を Studio キャンバスの入力元の右にドラッグします。

Anypoint Studio の右側にある [Mule Palette (Mule パレット)] ビューから MQTT Connector のいずれかを選択する

コネクタのグローバル要素を設定する

コネクタを設定する場合、アプリケーション内のそのコネクタのすべてのインスタンスで使用できるグローバル要素を設定することをお勧めします。MQTT Connector にはメッセージのパブリッシュ用とコンシューム用の両方のデフォルト値が用意されています。コネクタを使用するための唯一の要件として、使用する接続を設定し、その接続を一意に識別するクライアント ID の値を指定する必要があります。

クライアント ID はブローカーへの MQTT 接続を識別するため必須です。MQTT ブローカーに接続するクライアントまたはデバイスを一意に識別するわかりやすい名前 (ランダムな文字列ではない) を定義します。

MQTT Connector のグローバル要素を設定する手順は、次のとおりです。

Studio キャンバスで操作を選択します。
操作の設定画面で、[Connector configuration (コネクタ設定)] 項目の横にあるプラス記号 (+) をクリックし、グローバル要素設定項目にアクセスします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で、この設定に指定する接続種別のいずれかを選択します。
- MQTT3 URL Connection (MQTT3 URL 接続)
- MQTT3 Fail-Over Connection (MQTT3 フェールオーバー接続)
- MQTT3 Form Connection (MQTT3 フォーム接続)

ブローカーへの MQTT 接続を識別する一意の直観的な [Client ID (クライアント ID)] を指定します (例: smart-bentley-123)。
ブローカーに接続するための [URL] を指定します (例: tcp://127.0.0.1:1883)。
[General (一般)] タブで接続情報 (ブローカーで必要なライブラリなど) も指定します。
[LWT] タブで、必要に応じて LWT (Last Will and Testament) メッセージを指定します。
[SSL/TLS] タブで、必要に応じて TLS 設定を指定します。
[Advanced (詳細)] タブで、必要に応じて接続の詳細オプションと再接続戦略を指定します。
[OK] をクリックして、ウィンドウを閉じます。

MQTT 3 Connector の [Global Element Properties (グローバル要素プロパティ)] ウィンドウ

[Configuration XML (設定 XML)] エディターで、<mqtt3:connection>、clientId、および url の設定は次のように記述されます。

<mqtt3:config name="MQTT_Config">
    <mqtt3:connection clientId="smart-bentley-123" url="tcp://127.0.0.1:1883" />
</mqtt3:config>xml

[MQTT3 Form Connection (MQTT3 フォーム接続)] を設定することで、接続の各 URL 項目を別々に指定することもできます。これにより、ブローカーへの接続を確立するためのプロトコル、ホスト、およびポートを指定できます。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[Create (作成)] をクリックします。
検索条件ボックスに「mqtt」と入力し、[MQTT3 Config (MQTT3 設定)] を選択します。
[OK] をクリックします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で [MQTT3 Form Connection (MQTT3 フォーム接続)] を選択します。
次の項目を設定します。
- Client id generator (クライアント ID ジェネレーター): Client id random suffix generator
- Client ID (クライアント ID): smart-bentley-123
- Username (ユーザー名): username
- Protocol (プロトコル): TCP (Default) (TCP (デフォルト))
- URL: 127.0.0.1
- Port (ポート): 1883
[OK] をクリックします。

「Global Element Properties (グローバル要素プロパティ)」ウィンドウでの MQTT 3 フォーム接続の設定

[Configuration XML (設定 XML)] エディターで、<mqtt3:form-connection>、protocol、host、port、clientId の設定は次のように記述されます。

<mqtt3:config name="MQTT3_Config">
    <mqtt3:form-connection username="username" password="testpass" host="127.0.0.1" >
        <mqtt3:client-id-generator >
            <mqtt3:client-id-random-suffix-generator clientId="smart-bentley-123" />
        </mqtt3:client-id-generator>
    </mqtt3:form-connection>
</mqtt3:config>xml

接続プロトコルの設定

MQTT では、ブローカーへの接続およびブローカーとの MQTT メッセージの交換に使用できるプロトコルがサポートされます。コネクタ設定の接続文字列で次のいずれかのプロトコルを設定できます。

LOCAL
SSL
TCP (デフォルト)
WS
WSS

認証用のログイン情報の設定

認証ログイン情報は省略可能ですが、必要に応じてユーザー名とパスワードを指定できます。

次の例では、基本的なユーザー名とパスワードを指定することで認証を設定します。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[Create (作成)] をクリックします。
検索条件ボックスに「mqtt」と入力し、[MQTT3 Config (MQTT3 設定)] を選択します。
[OK] をクリックします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で [MQTT3 URL Connection (MQTT3 URL 接続)] を選択します。
次の項目を設定します。
- Client id generator (クライアント ID ジェネレーター): Client id random suffix generator
- Client ID (クライアント ID): smart-bentley-123
- Username (ユーザー名): username
- Password (パスワード): passtest
- URL: tcp://127.0.0.1:1883"
[OK] をクリックします。

[Configuration XML (設定 XML)] エディターで、<mqtt3:connection>、username、および password の設定は次のように記述されます。

<mqtt3:config name="MQTT_Config">
    <mqtt3:connection username="username" password="passtest" url='tcp://127.0.0.1:1883"' >
        <mqtt3:client-id-generator >
            <mqtt3:client-id-random-suffix-generator clientId="smart-bentley-123" />
        </mqtt3:client-id-generator>
    </mqtt3:connection>
</mqtt3:config>xml

TLS コンテキストを設定し、クライアント証明書を指定して接続を認証することもできます。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[MQTT3 Config (MQTT3 設定)] を選択して [Edit (編集)] をクリックします。
[MQTT3 Config (MQTT3 設定)] で [SSL/TLS] をクリックします。
[TLS Context (TLS コンテキスト)] で、[Edit Inline (インライン編集)] を選択します。
[Trust Store Configuration (トラストストア設定)] で、次の項目を設定します。
- Path (パス): tls/truststore.jks
- Password (パスワード): racing
- 型: jks
[OK] をクリックします。

「Global Element Properties (グローバル要素プロパティ)」ウィンドウでの MQTT の TLS コンテキストの設定

[Configuration XML (設定 XML)] エディターで、<tls:context>、path、password、type の設定は次のように記述されます。

<mqtt3:config name="MQTT_TLS_Config">
    <mqtt3:connection username="username" password="passtest" url='tcp://127.0.0.1:1883"' >
        <tls:context >
            <tls:trust-store path="tls/truststore.jks" password="racing" type="jks" />
        </tls:context>
        <mqtt3:client-id-generator >
            <mqtt3:client-id-random-suffix-generator clientId="smart-bentley-123" />
        </mqtt3:client-id-generator>
    </mqtt3:connection>
</mqtt3:config>xml

フェールオーバーサーバーリストの設定

連携して動作する複数のブローカーで構成され、クライアントに複数の接続エンドポイントを提供する特定のデプロイメントスキーマがあります。クライアントが接続できる複数の使用可能なサーバーがある場合、2 つのシナリオが考えられます。各 MQTT サーバーが個別に動作するか、サーバーが連携して状態を共有する (クラスターモード) ことができます。この場合、再接続時の MQTT クライアントの動作を指定することが必要になる場合があります。

フェールオーバーサーバーリストを提供すると、コネクタは、提供されたいずれかのエンドポイントとの接続を正常に確立するまでリストを反復処理することができます。

次の例では、[MQTT3 Fail-Over Connection (MQTT3 フェールオーバー接続)] のフェールオーバーサーバーリストを設定します。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[Create (作成)] をクリックします。
検索条件ボックスに「mqtt」と入力し、[MQTT3 Config (MQTT3 設定)] を選択します。
[OK] をクリックします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で [MQTT3 Fail-Over Connection (MQTT3 フェールオーバー接続)] を選択します。
[Client ID (クライアント ID)] を smart-bentley-123 に設定します。
[Fail over servers (フェールオーバーサーバー)] で、プラス記号 (+) をクリックします。
次の項目を設定します。
- Protocol (プロトコル): TCP (Default) (TCP (デフォルト))
- Host (ホスト): 127.9.0.2
- Port (ポート): 1883
[Finish (完了)] をクリックします。
ステップ 7 ～ 9 を 2 回繰り返して、もう一方の新しいホストとポートの値を追加します。

「Global Element Properties (グローバル要素プロパティ)」ウィンドウでの MQTT 3 フェールオーバーサーバーリストの設定

[Configuration XML (設定 XML)] エディターで、<mqtt3:fail-over-connection> および <mqtt3:fail-over-url の設定は次のように記述されます。

<mqtt3:config name="MQTT_FailOver_Config">
    <mqtt3:fail-over-connection clientIDGenerator="smart-bentley-123" >
        <mqtt3:fail-over-servers >
            <mqtt3:fail-over-url host="127.9.0.2" />
            <mqtt3:fail-over-url host="127.0.0.3" port="1884" />
            <mqtt3:fail-over-url host="127.0.0.1" />
        </mqtt3:fail-over-servers>
    </mqtt3:fail-over-connection>
</mqtt3:config>xml

[Clean Session (セッションをクリーンアップ)] の値の設定

[MQTT3 Config (MQTT3 設定)] ウィンドウの [Clean session (セッションをクリーンアップ)] 項目を False に設定すると、ブローカーは次回接続するクライアントを記憶します。クライアントがオフラインになっている間は、クライアントが再接続されるまで、そのすべてのサブスクリプションが保存され、クライアントで受信する必要があるサービス品質 (QoS) 1 と 2 のメッセージも保存されます。

一部のブローカーでは、MQTT ブローカーのクラスタリングがサポートされ、各ノードで状態が共有されます。この場合、ノードのコネクタがオフラインになることがある場合は、[セッションをクリーンアップ] フラグを False に設定しておくと便利です。これにより、クライアントのサブスクリプションを認識している別のノードにクライアントを再接続できるため、オフライン中にコネクタで見逃した可能性があるすべてのメッセージが配信されます。

[Clean Session (セッションをクリーンアップ)] が True (Default) に設定されている場合、コネクタが切断されると、理由にかかわらずそのすべてのサブスクリプションが削除されるため、コネクタは再接続時に再度サブスクライブする必要があります。オフライン中に送信されたすべてのメッセージは失われます。

次の例では、[Clean Session (セッションをクリーンアップ)] を設定します。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[MQTT3 Config (MQTT3 設定)] を選択して [Edit (編集)] をクリックします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] の [Advanced (詳細)] タブで、[MQTT3 URL Connection (MQTT3 URL 接続)] を選択します。
次の項目を設定します。
- Clean session (セッションをクリーンアップ): False
[OK] をクリックします。

「Global Element Properties (グローバル要素プロパティ)」ウィンドウでの MQTT の「Clean session (セッションをクリーンアップ)」の設定

設定 XML エディターでは、cleanSession 設定は次のように記述されます。

<mqtt3:config name="MQTT_Config">
    <mqtt3:connection username="username" password="passtest" url='tcp://127.0.0.1:1883' >
        <mqtt3:client-id-generator >
            <mqtt3:client-id-random-suffix-generator clientId="smart-bentley-123" />
        </mqtt3:client-id-generator>
        <mqtt3:connection-options cleanSession="false" />
    </mqtt3:connection>
</mqtt3:config>xml

Enable File Persistence (ファイル永続性を有効化)

[MQTT3 Config (MQTT3 設定)] ウィンドウで [Enable file persistence (ファイル永続性を有効化)] 項目を True に設定すると、MQTT クライアントはその状態をファイルに永続的に保持できます。このファイルは、クライアントに含まれている可能性のある QoS 1 以上のアウトバウンドまたはインバウンドインフライトメッセージの保存に使用されます。対照的に、この項目を False (Default) に設定すると、クライアントの状態はメモリにのみ保存されるため、クラッシュが発生すると、クライアントはその状態を復元できません。

[Enable File Persistence (ファイル永続性を有効化)] 機能では、オブジェクトストアは使用されません。このファイル機能は、ドライバーで永続的なメッセージ用にファイルを使用することに基づいており、オンプレミスインストールでのみ使用できます。

次の例では、ファイル永続性を有効化します。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[MQTT3 Config (MQTT3 設定)] を選択して [Edit (編集)] をクリックします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で [MQTT3 URL Connection (MQTT3 URL 接続)] を選択します。
[Advanced (詳細)] タブで、次の項目を設定します。
- Clean session (セッションをクリーンアップ): False
- Enable file persistence (ファイル永続性を有効化): True
[Datastore (データストア)] を mqtt/store に設定し、永続的なストアを生成する場所を指定します。
[OK] をクリックします。

「Global Element Properties (グローバル要素プロパティ)」ウィンドウでの MQTT 3 の「Enable File Persistence (ファイル永続性を有効化)」設定

Figure 1. MQTT3 の [Enable File Persistence (ファイル永続性を有効化)] 設定

[Configuration XML (設定 XML)] エディターで、enableFilePersistence および dataStorePath の設定は次のように記述されます。

<mqtt3:config name="MQTT_Config">
    <mqtt3:connection username="username" password="passtest" url='tcp://127.0.0.1:1883' >
        <mqtt3:client-id-generator >
            <mqtt3:client-id-random-suffix-generator clientId="smart-bentley-123" />
        </mqtt3:client-id-generator>
        <mqtt3:connection-options cleanSession="false" />
        <mqtt3:file-persistence-options enableFilePersistence="true" />
    </mqtt3:connection>
</mqtt3:config>xml

動的クライアント ID を設定した場合、MQTT Connector は、クラッシュが発生したときに、持続されたファイルを復元できません。クラッシュ後にアプリケーションが再起動しても変更されないクライアント ID を設定する必要があります。

接続の詳細オプションの設定

MQTT Connector を使用すると、メッセージをコンシュームまたはパブリッシュするときに複数のデフォルトパラメーターを定義できます。この方法では、設定に関連するすべての操作のグローバルのデフォルト動作を定義できます。

次の例では、[Keep alive internal (キープアライブ間隔)] 項目と [Keep alive internal unit (キープアライブ間隔単位)] 項目を設定し、クライアントとブローカー間でメッセージの交換がないまま接続を存続できる最大時間を設定します。また、[Max in flight (最大インフライト)] 項目を設定し、許可される最大インフライトメッセージ数を示します。

Studio で、[Global Elements (グローバル要素)] タブに移動します。
[MQTT3 Config (MQTT3 設定)] を選択して [Edit (編集)] をクリックします。
[MQTT3 Config (MQTT3 設定)] ウィンドウの [Connection (接続)] で [MQTT3 URL Connection (MQTT3 URL 接続)] を選択します。
[Advanced (詳細)] タブを選択し、次の項目を設定します。
- Keep alive internal (キープアライブ間隔): 60
- Keep Alive internal Unit (キープアライブ間隔単位): SECONDS (Default) (秒 (デフォルト))
- Max in flight (最大インフライト): 60
[OK] をクリックします。

「Global Element Properties (グローバル要素プロパティ)」ウィンドウでの MQTT3 Connector の接続の詳細設定

[Configuration XML (設定 XML)] エディターで、keepAliveInterval、keepAliveIntervalUnit、および maxInFlight の設定は次のように記述されます。

<mqtt3:config name="MQTT_Config">
    <mqtt3:connection username="username" password="passtest" url='tcp://127.0.0.1:1883' >
        <mqtt3:client-id-generator >
            <mqtt3:client-id-random-suffix-generator clientId="smart-bentley-123" />
        </mqtt3:client-id-generator>
        <mqtt3:connection-options keepAliveInterval="60" maxInFlight="60"/>
    </mqtt3:connection>
</mqtt3:config>xml

アプリケーションログの表示

問題を確認するため、アプリケーションログを次の方法で表示できます。

アプリケーションを Anypoint Platform から実行している場合、出力は Anypoint Studio のコンソールウィンドウに表示される。
コマンドラインから Mule を使用してアプリケーションを実行している場合、アプリケーションログは OS コンソールに表示される。

アプリケーションのログファイル (log4j2.xml) でログファイルパスがカスタマイズされていない場合、デフォルトの場所 MULE_HOME/logs/<app-name>.log でアプリケーションログを表示することもできます。