TsgcWSPClient_AMQP speaks AMQP 0.9.1, the protocol RabbitMQ was designed around. It exposes the model as it really is: you open a channel, declare an exchange and a queue, bind them with a routing key, then publish and consume. Nothing is hidden behind a simplified facade, so the broker behaves the way its own documentation says it will.
Channels are named with a string of your choosing, so 'ch1' identifies the same channel across every call rather than a numeric id you have to track.
TsgcWSPClient_AMQP
AMQP 0-9-1
TCP (5672) or TLS (5671)
Delphi, C++ Builder
Transport: plain TCP and TLS. sgcMQ connects over plain TCP and TLS. If you need AMQP over WebSocket, that requires a sgcWebSockets package, which provides the WebSocket client.
Do the declarations inside OnAMQPConnect, once the handshake has finished. Deliveries then arrive on 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}");Nearly every method has a blocking ...Ex twin that waits for the broker's reply and returns it, which is what you want in a linear setup routine.
OpenChannel and CloseChannel multiplex several logical conversations over one TCP connection. EnableChannel and DisableChannel apply channel flow control.
DeclareExchange(channel, name, type) creates an exchange of type direct, fanout, topic or headers. DeleteExchange removes it.
DeclareQueue, BindQueue, UnBindQueue, PurgeQueue and DeleteQueue cover the whole queue lifecycle, each with an ...Ex variant that returns the broker's response.
Consume(channel, queue) starts a subscription and CancelConsume ends it. Deliveries arrive on OnAMQPBasicDeliver, with OnAMQPBasicGetOk and OnAMQPBasicGetEmpty for single-message pulls.
PublishMessage(channel, exchange, routingKey, body) has overloads for text, streams and a full message object with headers and properties. Unroutable messages come back on OnAMQPBasicReturn.
AckMessage confirms a delivery tag, RejectMessage refuses it with an optional requeue, and Recover asks the broker to redeliver everything still unacknowledged.
SelectTransaction puts a channel into transactional mode, then CommitTransaction or RollbackTransaction closes it. OnAMQPTransactionOk confirms each step.
AMQPOptions carries VirtualHost, MaxChannels, MaxFrameSize and Locale, all negotiated with the broker during the opening handshake.
SetQoS(channel, prefetchSize, prefetchCount, global) caps how many unacknowledged messages the broker may push at a consumer, so a slow handler cannot be flooded.
HeartBeat keeps idle connections alive through proxies and past broker timeouts, with OnAMQPHeartBeat raised on each exchange.