このパッケージには 7 つのネイティブメッセージングクライアントが含まれています。MQTT が最短で動作させられるため、このページでは、ブローカーに接続し、トピックをサブスクライブし、メッセージをパブリッシュして、それが戻ってくるのを確認します。AMQP、Kafka、STOMP も同じ 2 コンポーネントの構成です。
ここでつまずく人が多い部分です。プロトコルコンポーネントが MQTT を話し、別のクライアントコンポーネントがソケットを管理します。両者は 1 回の代入で接続します。
TsgcWSPClient_MQTT は SGC MQ パレットページにあります。Connect、Subscribe、Publish、UnSubscribe と 13 個の MQTT イベントを備えています。
WebSocket 経由の MQTT には TsgcWebSocketClient、TCP 上の通常の MQTT には TsgcTCPClient を使用します。こちらに Host と Port を設定します。
MQTT.Client := WSClient;。その後、プロトコルのメソッドを呼び出すのではなく、WSClient.Active := True を設定して接続を開きます。
sgcWebSocket_Protocol_MQTT_Client.pas にはプラットフォームガードがなく、条件付きの Windows インポートがあるだけで、コンポーネントは ComponentPlatforms(0) で登録されています。すべてのターゲットでコンパイルできます。
エディションの列は、各プロトコルを制御する define と、それが Source/sgcVer.inc の何行目にあるかを示しています。
| 項目 | 内容 |
|---|---|
| IDE | Delphi 7 から RAD Studio 13、および C++Builder 2007 から 13。 |
| uses 句 | TsgcWSPClient_MQTT には sgcWebSocket_Protocols、クライアントには sgcWebSocket、TsgcWSConnection には sgcWebSocket_Classes、TmqttQoS には sgcWebSocket_Types を使用します。 |
| MQTT のエディション | SGC_MQTT は 717 行目、675 行目から 724 行目までの {$IFDEF SGC_EDT_STD} ブロック内で定義されています。つまり Standard 以上です。ブロックは累積式なので、Professional、Enterprise、All-Access にも含まれます。 |
| その他のプロトコル | すべて同じ Standard ブロック内にあります。SGC_STOMP は 716 行目、SGC_AMQP は 718 行目、SGC_AMQP1 は 719 行目、SGC_KAFKA は 720 行目、SGC_WAMP は 715 行目です。Enterprise の機能であるものはありません。 |
| エディション（単独パッケージ） | sgcMQ 製品は 855 行目で SGC_PACK_MQ を定義し、959 行目から 966 行目の独自ブロックで、同じ 5 つのプロトコルの define を有効にします。 |
| プラットフォーム | MQTT クライアントユニットにはユニットスコープのプラットフォームガードがありません。uses 句にある唯一の条件は {$IFDEF MSWINDOWS}Windows, {$ENDIF} です。 |
自分でブローカーをインストールせずにテストしたい場合は、同梱の MQTT デモが www.esegece.com の公開テストブローカーを指しており、WebSocket 経由の MQTT にはポート 15675 を使用します。
sgcMQ は sgcWebSockets のインストーラーに含まれており、単独のパッケージとしても提供されています。どちらでもインストールの手順は同じです。
ダウンロードしたファイルをフォルダーに展開します。以下ではこのフォルダーを {$DIR} と呼びます。
ツール、オプション、ライブラリの順に開きます。{$DIR}\source と、お使いの IDE 用の lib フォルダー（たとえば {$DIR}\libD13\$(Platform)）を追加します。
{$DIR}\Packages\ の下で、お使いの IDE バージョン用のパッケージグループを開きます。最初に実行時の .dpk をコンパイルし、次にデザイン時の dcl をインストールします。
SGC MQ というページが表示され、お使いのエディションで有効なプロトコルクライアントが含まれます。Standard ビルド以上では TsgcWSPClient_MQTT があります。
{$DIR}\Demos\02.WebSocket_Protocols\08.MQTT_Client を開いてください。以下のコードはここから抜粋したもので、すでに稼働中のテストブローカーを指しています。
プロトコルをクライアントに接続し、接続を開き、トピックをサブスクライブして、そこにパブリッシュします。メッセージは OnMQTTPublish で戻ってきます。
uses
Classes, SysUtils,
// sgc
sgcWebSocket, sgcWebSocket_Classes, sgcWebSocket_Protocols,
sgcWebSocket_Types, sgcWebSocket_Protocol_MQTT_Client,
sgcWebSocket_Protocol_MQTT_Message;
procedure TfrmClientPROTOCOL.FormCreate(Sender: TObject);
begin
WSClient := TsgcWebSocketClient.Create(Self);
WSClient.Host := 'www.esegece.com';
WSClient.Port := 15675;
WSClient.TLS := False;
// ... attach the client to the MQTT protocol
MQTT.Client := WSClient;
// ... and THIS is what opens the connection.
// There is no Connect call on the protocol component.
WSClient.Active := True;
end;WebSocket ではなく TCP 上の通常の MQTT を使う場合は、代わりに sgcTCP_Client_WS の TsgcTCPClient を作成し、MQTT.Client をそちらに向けます。他は何も変わりません。2 つのコンポーネントが分かれているのはそのためです。
procedure TfrmClientPROTOCOL.btnSubscribeClick(Sender: TObject);
begin
MQTT.Subscribe('sgc/quickstart');
end;
procedure TfrmClientPROTOCOL.btnPublishClick(Sender: TObject);
begin
// topic, payload, QoS, retain
MQTT.Publish('sgc/quickstart', 'hello from Delphi',
mtqsAtLeastOnce, False);
end;
procedure TfrmClientPROTOCOL.btnUnsubscribeClick(Sender: TObject);
begin
MQTT.UnSubscribe('sgc/quickstart');
end;Publish はパケット識別子である Word を返し、待機しません。PublishAndWait は Boolean を返し、ブローカーが確認するまでブロックします。タイムアウトの既定値は 10 秒です。TmqttQoS は sgcWebSocket_Types にあります。
procedure TfrmClientPROTOCOL.MQTTMQTTConnect(Connection: TsgcWSConnection;
const Session: Boolean; const ReasonCode: Integer;
const ReasonName: String;
const ConnectProperties: TsgcWSMQTTCONNACKProperties);
begin
memoLog.Lines.Add('#connected: ' + ReasonName);
end;
procedure TfrmClientPROTOCOL.MQTTMQTTSubscribe(Connection: TsgcWSConnection;
aPacketIdentifier: Word; aCodes: TsgcWSSUBACKS;
SubscribeProperties: TsgcWSMQTTSUBACKProperties);
begin
memoLog.Lines.Add('#subscribed: ' + IntToStr(aPacketIdentifier));
end;
procedure TfrmClientPROTOCOL.MQTTMQTTPublish(Connection: TsgcWSConnection;
aTopic, aText: string;
PublishProperties: TsgcWSMQTTPUBLISHProperties);
begin
memoLog.Lines.Add(aTopic + ': ' + aText);
end;最初に重要なのはこの 3 つです。確認応答のフロー、切断、MQTT 5.0 の認証用に、さらに 9 つあります。パラメーターリストが長く、イベントごとにプロパティオブジェクトが異なるため、ハンドラーは IDE に生成させてください。
3 つのタブはすべて、同梱のデモ Demos\02.WebSocket_Protocols\08.MQTT_Client\uClientMQTT.pas から、フォームコントロールをリテラルに置き換えたものです。そのプロジェクトでは MQTT オブジェクトをフォームに配置していますが、コードで作成しても同じように動作します。
2 つのイベントで、ブローカーが接続を受け入れ、トピックが有効になったことが分かります。
OnMQTTConnectブローカーがセッションを受け入れました。受け入れなかった場合は、ReasonCode と ReasonName でその理由が分かります。ソケットが切れるだけよりはるかに有用です。
OnMQTTSubscribeサブスクリプションが許可されました。TsgcWSSUBACKS である aCodes にはブローカーのトピックごとの応答が含まれるため、拒否されたサブスクリプションも通知されずに終わることはありません。
OnMQTTPublishメッセージが届きました。サブスクライブしたばかりのトピックにパブリッシュすると、そのまま戻ってきます。1 回のクリックで経路全体を確認できます。
QoS 1 と 2 では、配信は後から OnMQTTPubAck、OnMQTTPubRec、OnMQTTPubRel、OnMQTTPubComp で確認されます。代わりにブロッキング呼び出しが必要な場合は、PublishAndWait を使用します。
最初の接続が失敗する原因のほとんどは、6 つの問題のいずれかです。
接続がまだ開いていません。ソケットを開くのはプロトコルコンポーネントではなく、クライアントです。WSClient.Active := True を設定し、OnMQTTConnect を待ってからサブスクライブしてください。
MQTT.Client が一度も代入されていません。その 1 行がプロトコルをトランスポートに結び付けるもので、これがないとプロトコルには書き込み先がありません。
WebSocket 経由の MQTT と通常の TCP 上の MQTT は、ほとんどのブローカーでポートが異なります。TCP では通常 1883 と 8883 を使用し、WebSocket のエンドポイントはまったく別の場所にあります。同梱のデモは WebSocket に 15675 を使用します。
そのようなイベントはありません。ブローカーから受信したメッセージは、テキストの場合は OnMQTTPublish、生のペイロードと完全なメッセージオブジェクトが必要な場合は OnMQTTPublishEx で届きます。
確認応答は非同期で行われます。Publish はパケット識別子をすぐに返し、確認応答は後から OnMQTTPubAck、OnMQTTPubRec、OnMQTTPubRel、OnMQTTPubComp で届きます。ブロッキング呼び出しが必要な場合は PublishAndWait を使用してください。
ほとんどのブローカーはクライアント識別子と認証情報を要求します。Authentication に設定し、接続時にクライアント識別子を計算する必要がある場合は OnMQTTBeforeConnect を使用してください。
同じ 2 コンポーネントの構成が、パッケージ内の他のすべてのプロトコルでも使えます。
MQTTVersion を設定すると、リクエストとレスポンスのプロパティオブジェクトが利用可能になります。ユーザープロパティ、サブスクリプション識別子、トピックエイリアス、理由コードがすべて公開されます。
2 つの独立したクライアントがあり、0.9.1 は RabbitMQ 方式のエクスチェンジとキュー向け、1.0 は新しいワイヤープロトコル向けです。どちらも MQTT と同じ方法でクライアントに接続します。
ネイティブの Kafka ワイヤープロトコルクライアントと、RabbitMQ と ActiveMQ 向けのブローカー固有のバリアントを備えた STOMP 1.0 から 1.2 です。
AWS IoT と Azure IoT はどちらも MQTT を使用しており、ライブラリには署名と認証を処理する専用コンポーネントが含まれています。
リファレンスページには、すべてのプロパティとイベントが記載されています。デモプロジェクトはダウンロードに含まれており、Demos\02.WebSocket_Protocols の下にあります。
リファレンス、MQTT クライアント
TsgcWSPClient_MQTT のすべてのメソッド、プロパティ、イベント。
|
開く | |
| リファレンス、AMQP 0.9.1 エクスチェンジ、キュー、バインディング、コンシューマー。 | 開く | |
| リファレンス、AMQP 1.0 リンク、セッション、1.0 のメッセージモデル。 | 開く | |
| TsgcWSPClient_MQTT コンポーネントのページ MQTT クライアントの詳細と、他の 6 つのプロトコルクライアントへのリンクです。 | 開く | |
| プロトコルの概要 サブプロトコルコンポーネントをクライアントやサーバーに接続する方法です。 | 開く | |
| オンラインヘルプ 自動生成されたリファレンスで、常に最新リリースに対応しています。 | 開く |
関連資料：パブリッシュして確認応答を待つ、AMQP のエクスチェンジとキュー、Kafka クライアント。各製品にはそれぞれクイックスタートがあり、はじめにのページに一覧があります。
TsgcWSPClient_MQTT は MQTT パケットのエンコードとデコードを行います。TsgcWebSocketClient または TsgcTCPClient が、ソケット、ホスト、ポート、TLS の設定を管理します。両者は 1 回の代入 MQTT.Client := WSClient; で接続されます。
Active := True を設定します。同梱のデモはまさにそうしています。プロトコルにも Connect メソッドがありますが、デモでは使用しておらず、すべてのサンプルが従っているのは Active を設定する方法です。
SGC_MQTT は sgcVer.inc の 717 行目で、675 行目から 724 行目までの SGC_EDT_STD ブロック内で定義されています。これは有償で最も低い Standard エディションで、ブロックは累積式なので、より上位のすべてのエディションにも含まれます。716 行目の STOMP、718 行目の AMQP、719 行目の AMQP 1.0、720 行目の Kafka はすべて同じブロック内にあります。Enterprise の機能であるメッセージングプロトコルはありません。
OnMQTTPublish で、そのシグネチャは procedure(Connection: TsgcWSConnection; aTopic, aText: String; PublishProperties: TsgcWSMQTTPUBLISHProperties) です。OnMQTTMessage というイベントはありません。文字列ではなく生のバイトやメッセージオブジェクト全体が必要な場合は、代わりに OnMQTTPublishEx を処理してください。
Publish はパケット識別子である Word を返します。配信は非同期で確認されます。QoS 1 では OnMQTTPubAck、QoS 2 では OnMQTTPubRec、OnMQTTPubRel、OnMQTTPubComp を通じて確認されます。ブローカーが確認するまでブロックする呼び出しが必要な場合は、Boolean を返し、既定値 10 秒のタイムアウトを取る PublishAndWait を使用します。
Publish の 3 番目のパラメーターは TmqttQoS で、sgcWebSocket_Types.pas で、メンバー mtqsAtMostOnce、mtqsAtLeastOnce、mtqsExactlyOnce、mtqsReserved とともに宣言されています。既定値は mtqsAtMostOnce です。Subscribe も 2 番目のパラメーターに同じ型を取ります。
sgcWebSocket_Protocol_MQTT_Client.pas にはユニットスコープのプラットフォームガードがなく、uses 句にある唯一の条件は Windows のインポートで、パレットコンポーネントは ComponentPlatforms(0) で登録されているため、IDE がターゲットプラットフォームを制限することはありません。
MQTTVersion を設定します。すると、5.0 のプロパティオブジェクトが有効になります。送信時には ConnectProperties、戻りの各イベントにはプロパティのパラメーターがあり、たとえば OnMQTTConnect では TsgcWSMQTTCONNACKProperties、OnMQTTPublish では TsgcWSMQTTPUBLISHProperties です。