AMQP 1.0 Client
TsgcWSPClient_AMQP1은 ISO/IEC 19464로도 발행된 OASIS AMQP 1.0을 구현합니다. 이름은 같지만 AMQP 0.9.1과는 다른 프로토콜이며 객체 모델도 다릅니다. 컨테이너가 세션을 열고, 세션이 송신 링크와 수신 링크를 담으며, 링크가 크레딧 기반 흐름 제어 아래에서 메시지를 옮깁니다. Azure Service Bus와 Event Hubs가 사용하는 프로토콜이기도 합니다.
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이 이를 정리합니다. 세션은 두 컨테이너 사이의 순서와 윈도우를 관리하는 단위입니다.
CreateSenderLink는 대상 노드에, CreateReceiverLink는 소스 노드에 링크를 붙이고, CloseLink는 어느 쪽이든 분리합니다. 한 세션은 MaxLinksPerSession까지 여러 링크를 가질 수 있습니다.
SendMessage(session, link, text)가 메시지를 전송하며, 애플리케이션 속성이나 id, 주석이 필요할 때는 전체 TsgcAMQP1Message를 받는 오버로드를 사용합니다.
연결된 수신 링크는 전달 메시지를 OnAMQPMessage로 밀어 넣습니다. 블로킹 방식이 더 알맞을 때는 GetMessage가 타임아웃을 두고 하나를 동기적으로 가져옵니다.
OnAMQPMessageSent는 전송이 클라이언트를 떠날 때, OnAMQPMessageSentAck는 브로커가 이를 정산할 때 발생합니다. 두 이벤트는 의도적으로 분리되어 있습니다.
AMQPOptions.Authentication.AuthType으로 amqp1authSASLPlain, amqp1authSASLAnonymous, amqp1authSASLExternal 또는 인증 없음을 선택합니다. OnAMQPSASLAuthentication이 협상 과정을 알려 줍니다.
CreateCBSLink와 PutCBSToken이 Claims-Based Security를 구현합니다. CreateAzureCbsSasToken은 Service Bus SAS 토큰을, CreateAzureCbsJWT는 OAuth bearer 토큰을 만듭니다.
AMQPOptions.CreditSize와 WindowSize는 수신자가 추가 크레딧을 발급하기 전에 얼마나 받아들일지를 정합니다. MaxFrameSize와 ChannelMax는 연결 시 협상됩니다.
OnAMQPBeforeReadFrame과 OnAMQPBeforeWriteFrame은 원시 프레임을 Handled 플래그와 함께 전달하므로, 브로커가 예상과 다르게 동작할 때 큰 도움이 됩니다.