TsgcWSPClient_AMQP1 は、ISO/IEC 19464 としても発行されている OASIS AMQP 1.0 を実装します。名前は似ていますが AMQP 0.9.1 とは別のプロトコルであり、オブジェクトモデルも異なります。コンテナがセッションを開き、セッションが送信リンクと受信リンクを保持し、リンクがクレジットベースのフロー制御のもとでメッセージを運びます。Azure Service Bus と Event Hubs が話すプロトコルでもあります。
セッションとリンクは自分で付けた名前で指定するため、'session1' や 'sender1' が以降のすべての呼び出しで同じオブジェクトを指します。
TsgcWSPClient_AMQP1
OASIS AMQP 1.0（ISO/IEC 19464）
TCP（5672）または TLS（5671）
Delphi、C++ Builder
トランスポートは素の TCP と TLS です。sgcMQ は素の TCP と TLS で接続します。AMQP over WebSocket が必要な場合は、WebSocket クライアントを提供する sgcWebSockets のパッケージが必要です。
送信リンクはブローカー上のターゲットノードを、受信リンクはソースノードを指します。受信リンクをアタッチすると、配信が OnAMQPMessage に届きます。
uses
sgcTCP_Client_WS, sgcWebSocket_Protocols,
sgcWebSocket_Protocol_AMQP1_Client, sgcAMQP1_Classes, sgcAMQP1_Frames;
var
TCPClient: TsgcTCPClient;
AMQP1: TsgcWSPClient_AMQP1;
begin
TCPClient := TsgcTCPClient.Create(nil);
TCPClient.Host := 'broker.example.com';
TCPClient.Port := 5672;
AMQP1 := TsgcWSPClient_AMQP1.Create(nil);
AMQP1.Client := TCPClient;
AMQP1.AMQPOptions.ContainerId := 'delphi-app';
AMQP1.AMQPOptions.Authentication.AuthType := amqp1authSASLPlain;
AMQP1.AMQPOptions.Authentication.Username := 'guest';
AMQP1.AMQPOptions.Authentication.Password := 'guest';
AMQP1.OnAMQPConnect := AMQPConnect;
AMQP1.OnAMQPMessage := AMQPMessage;
TCPClient.Active := True;
end;
procedure TForm1.AMQPConnect(Sender: TObject;
const aOpen: TsgcAMQP1FrameOpen);
begin
AMQP1.CreateSession('session1');
// Sender link: name, then the target node address
AMQP1.CreateSenderLink('session1', 'sender1', '/queue/orders');
// Receiver link: deliveries arrive on OnAMQPMessage
AMQP1.CreateReceiverLink('session1', 'receiver1', '/queue/orders');
AMQP1.SendMessage('session1', 'sender1', '{"id":42}');
end;#include "sgcTCP_Client_WS.hpp"
#include "sgcWebSocket_Protocols.hpp"
#include "sgcWebSocket_Protocol_AMQP1_Client.hpp"
TsgcTCPClient *TCPClient = new TsgcTCPClient(this);
TCPClient->Host = "broker.example.com";
TCPClient->Port = 5672;
TsgcWSPClient_AMQP1 *AMQP1 = new TsgcWSPClient_AMQP1(this);
AMQP1->Client = TCPClient;
AMQP1->AMQPOptions->ContainerId = "cbuilder-app";
AMQP1->OnAMQPConnect = AMQPConnect;
AMQP1->OnAMQPMessage = AMQPMessage;
TCPClient->Active = true;
// In the OnAMQPConnect handler. SendMessage is SendMessage_ in C++ Builder.
AMQP1->CreateSession("session1");
AMQP1->CreateSenderLink("session1", "sender1", "/queue/orders");
AMQP1->CreateReceiverLink("session1", "receiver1", "/queue/orders");
AMQP1->SendMessage_("session1", "sender1", "{\"id\":42}");最もよく使うメンバーです。
CreateSession は TsgcAMQP1Session を返し、CloseSession がそれを破棄します。セッションは、2 つのコンテナ間における順序制御とウィンドウ制御の単位です。
CreateSenderLink はターゲットノードに、CreateReceiverLink はソースノードにアタッチし、CloseLink はいずれもデタッチします。1 つのセッションは MaxLinksPerSession まで複数のリンクを保持できます。
SendMessage(session, link, text) がメッセージを転送します。アプリケーションプロパティ、ID、アノテーションが必要な場合は、完全な TsgcAMQP1Message を受け取るオーバーロードを使います。
アタッチ済みの受信リンクは、配信を OnAMQPMessage に送ります。ブロッキング形式のほうが適している場合は、GetMessage でタイムアウト付きに 1 件を同期的に取得できます。
OnAMQPMessageSent は転送がクライアントを離れた時点で、OnAMQPMessageSentAck はブローカーが確定した時点で発生します。両者は意図的に分けられています。
AMQPOptions.Authentication.AuthType で amqp1authSASLPlain、amqp1authSASLAnonymous、amqp1authSASLExternal、または認証なしを選択します。OnAMQPSASLAuthentication がネゴシエーションの結果を通知します。
CreateCBSLink と PutCBSToken が Claims-Based Security を実装します。CreateAzureCbsSasToken は Service Bus の SAS トークンを、CreateAzureCbsJWT は OAuth のベアラートークンを生成します。
AMQPOptions.CreditSize と WindowSize は、受信側が追加のクレジットを発行するまでにどれだけ受け入れるかを決めます。MaxFrameSize と ChannelMax は接続開始時にネゴシエートされます。
OnAMQPBeforeReadFrame と OnAMQPBeforeWriteFrame は、各生フレームを Handled フラグとともに渡します。ブローカーが想定外の動作をする場合に非常に役立ちます。