STOMP ActiveMQ クライアント
TsgcWSPClient_STOMP_ActiveMQ は、Apache ActiveMQ 向けに調整された STOMP クライアントです。ActiveMQ のデスティネーション命名規則に沿ったキューとトピックのヘルパー、すべてのイベントで受け取れる型付きヘッダーオブジェクト、そして永続サブスクリプションに必要なブローカー固有の設定を扱う ActiveMQ_Options プロパティが追加されています。
TsgcWSPClient_STOMP_ActiveMQ は、Apache ActiveMQ 向けに調整された STOMP クライアントです。ActiveMQ のデスティネーション命名規則に沿ったキューとトピックのヘルパー、すべてのイベントで受け取れる型付きヘッダーオブジェクト、そして永続サブスクリプションに必要なブローカー固有の設定を扱う ActiveMQ_Options プロパティが追加されています。
汎用 STOMP クライアントの内容はすべてそのまま当てはまります。追加されるのは、デスティネーションのヘルパー、ActiveMQ 名のイベント、ブローカーのオプションです。
TsgcWSPClient_STOMP_ActiveMQ
STOMP 1.0, 1.1 and 1.2
TCP (61613) or TLS (61612)
Delphi、C++ Builder
トランスポートは素の TCP と TLS です。sgcMQ は素の TCP と TLS で ActiveMQ と通信します。ActiveMQ の STOMP-over-WebSocket コネクターは WebSocket エンドポイントなので、接続するには WebSocket クライアントを提供する sgcWebSockets のパッケージが必要です。
このサンプルは、ActiveMQ のネイティブ STOMP ポート 61613 を利用しています。暗号化した接続にするには、キャリア側で TLS を有効にし、ポート 61612 のブローカーの stomp+ssl コネクターを使ってください。
uses
sgcTCP_Client_WS, sgcWebSocket_Classes, sgcWebSocket_Protocols,
sgcWebSocket_Protocol_STOMP_Broker_Client,
sgcWebSocket_Protocol_STOMP_ActiveMQ_Client;
var
TCPClient: TsgcTCPClient;
STOMP: TsgcWSPClient_STOMP_ActiveMQ;
begin
TCPClient := TsgcTCPClient.Create(nil);
TCPClient.Host := 'broker.example.com';
TCPClient.Port := 61613;
STOMP := TsgcWSPClient_STOMP_ActiveMQ.Create(nil);
STOMP.Client := TCPClient;
STOMP.ActiveMQ_Options.ClientId := 'delphi-consumer-1';
STOMP.Authentication.Enabled := True;
STOMP.Authentication.UserName := 'admin';
STOMP.Authentication.Password := 'admin';
STOMP.OnActiveMQConnected := ActiveMQConnected;
STOMP.OnActiveMQMessage := ActiveMQMessage;
TCPClient.Active := True;
end;
procedure TForm1.ActiveMQConnected(Connection: TsgcWSConnection;
Headers: TsgcWSActiveMQSTOMPHeadersConnected);
begin
STOMP.SubscribeQueue('orders');
STOMP.PublishQueue('orders', '{"id":42}');
end;
procedure TForm1.ActiveMQMessage(Connection: TsgcWSConnection;
MessageText: string; Headers: TsgcWSActiveMQSTOMPHeadersMessage;
Subscription: TsgcWSBrokerSTOMPSubscriptionItem);
begin
Memo1.Lines.Add(Headers.Destination + ': ' + MessageText);
end;// include: sgcTCP_Client_WS.hpp, sgcWebSocket_Protocols.hpp,
// sgcWebSocket_Protocol_STOMP_ActiveMQ_Client.hpp
TsgcTCPClient *TCPClient = new TsgcTCPClient(this);
TCPClient->Host = "broker.example.com";
TCPClient->Port = 61613;
TsgcWSPClient_STOMP_ActiveMQ *STOMP =
new TsgcWSPClient_STOMP_ActiveMQ(this);
STOMP->Client = TCPClient;
STOMP->ActiveMQ_Options->ClientId = "cbuilder-consumer-1";
STOMP->Authentication->Enabled = true;
STOMP->Authentication->UserName = "admin";
STOMP->Authentication->Password = "admin";
STOMP->OnActiveMQConnected = ActiveMQConnected;
STOMP->OnActiveMQMessage = ActiveMQMessage;
TCPClient->Active = true;
// In the OnActiveMQConnected handler:
STOMP->SubscribeQueue("orders");
STOMP->PublishQueue("orders", "{\"id\":42}");最もよく使うメンバーです。
SubscribeQueue、PublishQueue、UnSubscribeQueue は素のキュー名を受け取り、/queue/ デスティネーションを組み立てます。durable、exclusive、ack モードを引数で指定できます。
SubscribeTopic、PublishTopic、UnSubscribeTopic は /topic/ デスティネーションで同じことを行います。永続サブスクリプションが最も重要になる領域です。
SubscribeEx、PublishEx、UnSubscribeEx は生のデスティネーション文字列を受け取り、名前付きヘルパーでカバーできない用途に対応します。
ActiveMQ_Options は ActiveMQ 固有の設定を保持します。復帰した永続サブスクライバーをブローカーが識別するために使う ClientId もここに含まれます。
サブスクライブ用のヘルパーは Durable フラグを受け取ります。トピックのサブスクライバーは切断したあとで、ブローカーが保持していたメッセージをまとめて受け取れます。
OnActiveMQMessage は TsgcWSActiveMQSTOMPHeadersMessage と対応するサブスクリプション項目を渡すため、デスティネーション、メッセージ ID、ack モードをプロパティとして扱えます。
サブスクライブ時に ack モードを渡し、メッセージ ID を指定して ACK または NACK を呼び出します。ackAuto では、ブローカーは送信した時点でメッセージが配信されたとみなします。
各パブリッシュヘルパーは、コンテンツタイプ、任意のトランザクション名、そして優先度、永続化、有効期限といった ActiveMQ のヘッダー用のメッセージオプションオブジェクトを受け取ります。
Client、Authentication、HeartBeat、Versions、Options は、汎用 STOMP クライアントとまったく同じように動作します。