Delphi と C++Builder 向けの 2 つのネイティブ AMQP クライアントコンポーネント。TsgcWSPClient_AMQP はエクスチェンジ、キュー、バインディング、コンシューマー確認応答を備えて RabbitMQ と AMQP 0.9.1 で通信し、TsgcWSPClient_AMQP1 は WebSocket または生 TCP キャリア上で Azure Service Bus、Apache Qpid、その他あらゆる準拠ブローカーと OASIS AMQP 1.0 で通信します。
AMQP 0.9.1 と AMQP 1.0 は名前こそ同じですが、異なるワイヤープロトコルです。sgcWebSockets はそれぞれをネイティブに実装しているので、ブローカーに合ったコンポーネントを選ぶだけです。
Delphi AMQP クライアントは、VCL または FMX アプリケーションをエンタープライズメッセージブローカーに接続します。AMQP 0.9.1 は RabbitMQ が使用する、広くデプロイされたバージョンです: エクスチェンジとキューを宣言し、ルーティングキーでバインドし、basic.publish で発行、basic.consume で消費し、その上にトランザクションと confirm が乗ります。AMQP 1.0 は Azure Service Bus と Apache Qpid が使用する OASIS 標準です: セッション、センダーリンク、レシーバーリンクの上に構築された対称的なリンクプロトコルで、Azure Service Bus の SAS / OAuth 認証向けの Claims-Based-Security トークンヘルパーを備えます。
両コンポーネントとも Client プロパティで TsgcWebSocketClient にアタッチし、WebSocket または生 TCP キャリア上で動作します。同じコードで標準の AMQP ポート上のブローカーに到達することも、HTTP 専用インフラを越えてポート 443 でトンネルすることもできます。
TsgcWSPClient_AMQP: エクスチェンジ、キュー、バインディング、トランザクション
TsgcWSPClient_AMQP1: セッション、センダー／レシーバーリンク、CBS トークン
WebSocket または生 TCP キャリア、いずれも TLS は任意
Windows、macOS、Linux、iOS、Android。Delphi 7 から Delphi 13 まで
あらゆるプロトコル操作がメソッドに、あらゆるブローカー通知がイベントになります。
DeclareExchange、DeclareQueue、BindQueue がコードから RabbitMQ のルーティングトポロジーを構築します: direct、fanout、topic ルーティングと、OpenChannel によるチャネル単位のスコープ。
PublishMessage はルーティングキー付きでエクスチェンジに送信し、Consume はキューを購読して、配信は OnAMQPBasicDeliver 経由で到着します。
CreateSession が AMQP 1.0 セッションを開き、続いて CreateSenderLink と CreateReceiverLink がノードアドレスにアタッチします。受信転送は OnAMQPMessage 経由で到着します。
Claims-Based-Security ヘルパーが Azure Service Bus の SAS / OAuth トークンフローをラップするため、ネームスペースへの認証は手組みの CBS 交換ではなくプロパティ設定で済みます。
OnAMQPMessageSent と OnAMQPMessageSentAck が、転送がクライアントを離れた時点とブローカーがそれを確定した時点を報告します。
認証チャレンジは OnAMQPSASLAuthentication（1.0）と OnAMQPAuthentication / OnAMQPChallenge（0.9.1）を通じて表面化します。
0.9.1 クライアントは HeartBeat プロパティを公開しており、アイドル接続がプロキシやブローカーのタイムアウトを越えて生存し続けます。
OnAMQPConnect、OnAMQPDisconnect、OnAMQPClose、OnAMQPException がアプリケーションにプロトコル状態を通知し続けます。
両 AMQP クライアントは、Standard から Enterprise、All-Access まで sgcWebSockets のすべてのエディションに、さらに .NET ライブラリにも含まれます。
AMQP コンポーネントを TsgcWebSocketClient とペアにして接続し、宣言／アタッチしたら準備完了。
uses
sgcWebSocket, sgcWebSocket_Protocol_AMQP_Client, sgcAMQP_Classes;
var
WSClient: TsgcWebSocketClient;
AMQP: TsgcWSPClient_AMQP;
begin
WSClient := TsgcWebSocketClient.Create(nil);
WSClient.Host := 'broker.example.com';
WSClient.Port := 15674;
WSClient.Options.Parameters := '/ws';
AMQP := TsgcWSPClient_AMQP.Create(nil);
AMQP.Client := WSClient;
AMQP.OnAMQPConnect := AMQPConnect;
AMQP.OnAMQPBasicDeliver := AMQPBasicDeliver;
WSClient.Active := True;
end;
procedure TForm1.AMQPConnect(Sender: TObject);
begin
AMQP.OpenChannel('ch1');
AMQP.DeclareExchange('ch1', 'orders', 'direct');
AMQP.DeclareQueue('ch1', 'orders_in');
AMQP.BindQueue('ch1', 'orders_in', 'orders', 'create');
AMQP.Consume('ch1', 'orders_in');
// Publish a message
AMQP.PublishMessage('ch1', 'orders', 'create', '{"id":42}');
end;uses
sgcWebSocket, sgcWebSocket_Protocol_AMQP1_Client, sgcAMQP1_Classes;
var
WSClient: TsgcWebSocketClient;
AMQP1: TsgcWSPClient_AMQP1;
oSession: TsgcAMQP1Session;
oSender: TsgcAMQP1SenderLink;
begin
WSClient := TsgcWebSocketClient.Create(nil);
WSClient.Host := 'broker.example.com';
WSClient.Port := 5673;
AMQP1 := TsgcWSPClient_AMQP1.Create(nil);
AMQP1.Client := WSClient;
AMQP1.OnAMQPConnect := AMQPConnect;
AMQP1.OnAMQPMessage := AMQPMessage;
AMQP1.OnAMQPSessionOpen := AMQPSessionOpen;
AMQP1.OnAMQPLinkOpen := AMQPLinkOpen;
WSClient.Active := True;
oSession := AMQP1.CreateSession;
oSender := AMQP1.CreateSenderLink(oSession, '/queue/orders');
// Receivers handle pushed deliveries through OnAMQPMessage
AMQP1.CreateReceiverLink(oSession, '/queue/orders');
end;#include "sgcWebSocket.hpp"
#include "sgcWebSocket_Protocol_AMQP_Client.hpp"
TsgcWebSocketClient* WSClient = new TsgcWebSocketClient(NULL);
WSClient->Host = "broker.example.com";
WSClient->Port = 15674;
WSClient->Options->Parameters = "/ws";
TsgcWSPClient_AMQP* AMQP = new TsgcWSPClient_AMQP(NULL);
AMQP->Client = WSClient;
AMQP->OnAMQPConnect = AMQPConnect;
AMQP->OnAMQPBasicDeliver = AMQPBasicDeliver;
WSClient->Active = true;
// In the OnAMQPConnect handler:
AMQP->OpenChannel("ch1");
AMQP->DeclareExchange("ch1", "orders", "direct");
AMQP->DeclareQueue("ch1", "orders_in");
AMQP->BindQueue("ch1", "orders_in", "orders", "create");
AMQP->Consume("ch1", "orders_in");
AMQP->PublishMessage("ch1", "orders", "create", "{\"id\":42}");バージョン番号ではなくブローカーで選んでください。相手が RabbitMQ なら AMQP 0.9.1 コンポーネントを使用します: エクスチェンジ、キュー、バインディングという RabbitMQ のモデルに 1 対 1 で対応します。相手が Azure Service Bus、Apache Qpid、あるいは OASIS 標準を掲げる他のブローカーなら、AMQP 1.0 コンポーネントとそのセッション／リンクモデルを使用してください。
両方とも同じパッケージで提供されるため、混在環境でも問題ありません: 1 つのアプリケーションが、それぞれ専用の TsgcWebSocketClient キャリア上で、RabbitMQ への 0.9.1 接続と Azure Service Bus への 1.0 接続を並べて保持できます。
TsgcWSPClient_AMQP の完全リファレンス: プロパティ、メソッド、イベント、デモプロジェクト。
TsgcWSPClient_AMQP1 の完全リファレンス: セッション、リンク、CBS 認証。
MQTT、WebSocket、SSE、HTTP/2 push よりも AMQP をいつ選ぶか。
Delphi コードで解説する AMQP 0.9.1 ルーティングモデル。
Delphi クライアントからの発行パターンと配信オプション。
AMQP 1.0 上での Azure Service Bus に対する Claims-Based-Security トークン。