TsgcWSPClient_STOMP_RabbitMQ は、RabbitMQ のデスティネーション規約を組み込んだ STOMP クライアントです。RabbitMQ はデスティネーション文字列に意味を埋め込むため、/queue/、/topic/、/exchange/、/amq/queue/、/temp-queue/ はそれぞれ別の意味を持ちます。本コンポーネントはこれらを名前付きメソッドに置き換え、生の文字列解析の代わりに型付きのヘッダーレコードを提供します。
汎用 STOMP クライアントの機能はすべてそのまま使えます。追加されるのは、デスティネーションのヘルパーと、型付きヘッダーオブジェクトを伴う RabbitMQ 名のイベントです。
TsgcWSPClient_STOMP_RabbitMQ
STOMP 1.0, 1.1 and 1.2
TCP（61613）または TLS（61614）
Delphi、C++ Builder
トランスポートは素の TCP と TLS です。sgcMQ は素の TCP と TLS で RabbitMQ の STOMP プラグインと通信します。RabbitMQ の Web-STOMP プラグインは WebSocket エンドポイントなので、接続するには WebSocket クライアントを提供する sgcWebSockets のパッケージが必要です。
このサンプルは、RabbitMQ の STOMP プラグインをネイティブの TCP ポート 61613 で利用しています。暗号化した接続にするには、キャリア側で TLS を有効にしてポート 61614 を使ってください。
uses
sgcTCP_Client_WS, sgcWebSocket_Classes, sgcWebSocket_Protocols,
sgcWebSocket_Protocol_STOMP_Broker_Client,
sgcWebSocket_Protocol_STOMP_RabbitMQ_Client;
var
TCPClient: TsgcTCPClient;
STOMP: TsgcWSPClient_STOMP_RabbitMQ;
begin
TCPClient := TsgcTCPClient.Create(nil);
TCPClient.Host := 'rabbit.example.com';
TCPClient.Port := 61613;
STOMP := TsgcWSPClient_STOMP_RabbitMQ.Create(nil);
STOMP.Client := TCPClient;
STOMP.Authentication.Enabled := True;
STOMP.Authentication.UserName := 'guest';
STOMP.Authentication.Password := 'guest';
STOMP.OnRabbitMQConnected := RabbitMQConnected;
STOMP.OnRabbitMQMessage := RabbitMQMessage;
TCPClient.Active := True;
end;
procedure TForm1.RabbitMQConnected(Connection: TsgcWSConnection;
Headers: TsgcWSRabbitMQSTOMPHeadersConnected);
begin
// /queue/orders, durable
STOMP.SubscribeQueue('orders');
// /exchange/orders with routing pattern 'create'
STOMP.SubscribeExchange('orders', 'create');
STOMP.PublishExchange('orders', 'create', '{"id":42}');
end;
procedure TForm1.RabbitMQMessage(Connection: TsgcWSConnection;
MessageText: string; Headers: TsgcWSRabbitMQSTOMPHeadersMessage;
Subscription: TsgcWSBrokerSTOMPSubscriptionItem);
begin
Memo1.Lines.Add(Headers.Destination + ': ' + MessageText);
end;// include: sgcTCP_Client_WS.hpp, sgcWebSocket_Protocols.hpp,
// sgcWebSocket_Protocol_STOMP_RabbitMQ_Client.hpp
TsgcTCPClient *TCPClient = new TsgcTCPClient(this);
TCPClient->Host = "rabbit.example.com";
TCPClient->Port = 61613;
TsgcWSPClient_STOMP_RabbitMQ *STOMP =
new TsgcWSPClient_STOMP_RabbitMQ(this);
STOMP->Client = TCPClient;
STOMP->Authentication->Enabled = true;
STOMP->Authentication->UserName = "guest";
STOMP->Authentication->Password = "guest";
STOMP->OnRabbitMQConnected = RabbitMQConnected;
STOMP->OnRabbitMQMessage = RabbitMQMessage;
TCPClient->Active = true;
// In the OnRabbitMQConnected handler:
STOMP->SubscribeQueue("orders");
STOMP->SubscribeExchange("orders", "create");
STOMP->PublishExchange("orders", "create", "{\"id\":42}");各ヘルパーは素の名前を受け取り、RabbitMQ のデスティネーション文字列はコンポーネントが組み立てます。
SubscribeQueue、PublishQueue、UnSubscribeQueue は RabbitMQ の /queue/ デスティネーションを扱い、durable、auto-delete、exclusive の各フラグを引数で指定できます。
SubscribeTopic、PublishTopic、UnSubscribeTopic は /topic/ デスティネーションを使い、RabbitMQ の amq.topic エクスチェンジ経由でルーティングされます。
SubscribeExchange(name, pattern) と PublishExchange(name, routingKey, text) は /exchange/ デスティネーションを指定します。バインディングのパターンで受信内容を選択します。
SubscribeQueueOutside と PublishQueueOutside は /amq/queue/ に対応します。別の場所で宣言済みで、クライアントが再宣言してはならないキュー向けです。
SubscribeTemporaryQueue(queue, replyTo) と PublishTemporaryQueue は、短命な返信先デスティネーションを使う RabbitMQ の /temp-queue/ リクエスト／レスポンスパターンを実装します。
SubscribeEx、PublishEx、UnSubscribeEx は生のデスティネーション文字列を受け取ります。名前付きヘルパーでカバーできない場合に使います。
OnRabbitMQMessage は TsgcWSRabbitMQSTOMPHeadersMessage と対応するサブスクリプション項目を渡します。デスティネーション、メッセージ ID、ack モードが部分文字列ではなくプロパティとして扱えます。
各パブリッシュヘルパーは追加ヘッダーの TStrings を受け取ります。メッセージの優先度、有効期限、永続化といった RabbitMQ の拡張はここで設定します。
Client、Authentication、HeartBeat、Versions、Options は、汎用 STOMP クライアントとまったく同じように動作します。