TsgcWSPClient_AMQP spreekt AMQP 0.9.1, het protocol waar RabbitMQ omheen is ontworpen. Het laat het model zien zoals het echt is: je opent een kanaal, declareert een exchange en een wachtrij, bindt ze met een routing key en publiceert en consumeert daarna. Er wordt niets verborgen achter een vereenvoudigde façade, dus de broker gedraagt zich precies zoals zijn eigen documentatie zegt.
Kanalen krijgen een naam die je zelf kiest, dus 'ch1' verwijst bij elke aanroep naar hetzelfde kanaal in plaats van naar een numerieke id die je moet bijhouden.
TsgcWSPClient_AMQP
AMQP 0-9-1
TCP (5672) of TLS (5671)
Delphi, C++ Builder
Transport: gewoon TCP en TLS. sgcMQ verbindt via gewoon TCP en TLS. Heb je AMQP over WebSocket nodig, dan is daarvoor een sgcWebSockets-pakket vereist, dat de WebSocket-client levert.
Doe de declaraties binnen OnAMQPConnect, zodra de handshake klaar is. Afleveringen komen daarna binnen op OnAMQPBasicDeliver.
uses
sgcTCP_Client_WS, sgcWebSocket_Protocols,
sgcWebSocket_Protocol_AMQP_Client, sgcAMQP_Classes;
var
TCPClient: TsgcTCPClient;
AMQP: TsgcWSPClient_AMQP;
begin
TCPClient := TsgcTCPClient.Create(nil);
TCPClient.Host := 'broker.example.com';
TCPClient.Port := 5672;
AMQP := TsgcWSPClient_AMQP.Create(nil);
AMQP.Client := TCPClient;
AMQP.AMQPOptions.VirtualHost := '/';
AMQP.HeartBeat.Enabled := True;
AMQP.HeartBeat.Interval := 30;
AMQP.OnAMQPConnect := AMQPConnect;
AMQP.OnAMQPBasicDeliver := AMQPBasicDeliver;
TCPClient.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 to the exchange with a routing key
AMQP.PublishMessage('ch1', 'orders', 'create', '{"id":42}');
end;#include "sgcTCP_Client_WS.hpp"
#include "sgcWebSocket_Protocols.hpp"
#include "sgcWebSocket_Protocol_AMQP_Client.hpp"
TsgcTCPClient *TCPClient = new TsgcTCPClient(this);
TCPClient->Host = "broker.example.com";
TCPClient->Port = 5672;
TsgcWSPClient_AMQP *AMQP = new TsgcWSPClient_AMQP(this);
AMQP->Client = TCPClient;
AMQP->OnAMQPConnect = AMQPConnect;
AMQP->OnAMQPBasicDeliver = AMQPBasicDeliver;
TCPClient->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}");Bijna elke methode heeft een blokkerende ...Ex-tegenhanger die op het antwoord van de broker wacht en dat teruggeeft, precies wat je wilt in een lineaire setuproutine.
OpenChannel en CloseChannel multiplexen meerdere logische conversaties over één TCP-verbinding. EnableChannel en DisableChannel passen flow control op kanaalniveau toe.
DeclareExchange(channel, name, type) maakt een exchange van het type direct, fanout, topic of headers. DeleteExchange verwijdert hem.
DeclareQueue, BindQueue, UnBindQueue, PurgeQueue en DeleteQueue dekken de volledige levenscyclus van een wachtrij, elk met een ...Ex-variant die het antwoord van de broker teruggeeft.
Consume(channel, queue) start een abonnement en CancelConsume beëindigt het. Afleveringen komen binnen op OnAMQPBasicDeliver, met OnAMQPBasicGetOk en OnAMQPBasicGetEmpty voor het ophalen van losse berichten.
PublishMessage(channel, exchange, routingKey, body) heeft overloads voor tekst, streams en een volledig berichtobject met headers en eigenschappen. Berichten die niet gerouteerd kunnen worden, komen terug op OnAMQPBasicReturn.
AckMessage bevestigt een delivery tag, RejectMessage weigert hem met een optionele requeue, en Recover vraagt de broker om alles wat nog niet bevestigd is opnieuw af te leveren.
SelectTransaction zet een kanaal in transactiemodus, daarna sluit CommitTransaction of RollbackTransaction hem af. OnAMQPTransactionOk bevestigt elke stap.
AMQPOptions bevat VirtualHost, MaxChannels, MaxFrameSize en Locale, die allemaal tijdens de openingshandshake met de broker worden onderhandeld.
SetQoS(channel, prefetchSize, prefetchCount, global) begrenst hoeveel onbevestigde berichten de broker naar een consumer mag sturen, zodat een trage handler niet overspoeld raakt.
HeartBeat houdt inactieve verbindingen in leven door proxy's heen en voorbij de time-outs van de broker, waarbij OnAMQPHeartBeat bij elke uitwisseling wordt afgevuurd.
| Online helpVolledige API-referentie en gebruikershandleiding. | Openen | |
| AMQP 1.0-clientDe andere AMQP, een ander protocol met een eigen component. | Openen | |
| Download de gratis proefversieDraai de client tegen een lokale RabbitMQ-container. | Openen | |
| PrijzenSingle-, Team- en Site-licenties met volledige broncode. | Openen |