TsgcWSPClient_AMQP parla AMQP 0.9.1, il protocollo attorno al quale RabbitMQ è stato progettato. Espone il modello per quello che è davvero: apri un canale, dichiari un exchange e una coda, li colleghi con una routing key, poi pubblichi e consumi. Niente è nascosto dietro una facciata semplificata, quindi il broker si comporta come dice la sua stessa documentazione.
I canali hanno un nome scelto da te come stringa, quindi 'ch1' identifica lo stesso canale in ogni chiamata invece di un id numerico da tenere sotto controllo.
TsgcWSPClient_AMQP
AMQP 0-9-1
TCP (5672) o TLS (5671)
Delphi, C++ Builder
Trasporto: TCP semplice e TLS. sgcMQ si connette tramite TCP semplice e TLS. Se ti serve AMQP su WebSocket, è necessario un package sgcWebSockets, che fornisce il client WebSocket.
Fai le dichiarazioni dentro OnAMQPConnect, una volta terminato l'handshake. Le consegne arrivano poi su 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}");Quasi ogni metodo ha un gemello bloccante ...Ex che attende la risposta del broker e la restituisce, che è quello che serve in una routine di configurazione lineare.
OpenChannel e CloseChannel multiplexano diverse conversazioni logiche su una sola connessione TCP. EnableChannel e DisableChannel applicano il controllo di flusso del canale.
DeclareExchange(channel, name, type) crea un exchange di tipo direct, fanout, topic o headers. DeleteExchange lo rimuove.
DeclareQueue, BindQueue, UnBindQueue, PurgeQueue e DeleteQueue coprono l'intero ciclo di vita della coda, ognuno con una variante ...Ex che restituisce la risposta del broker.
Consume(channel, queue) avvia una sottoscrizione e CancelConsume la termina. Le consegne arrivano su OnAMQPBasicDeliver, con OnAMQPBasicGetOk e OnAMQPBasicGetEmpty per il prelievo di un singolo messaggio.
PublishMessage(channel, exchange, routingKey, body) ha overload per testo, stream e un oggetto messaggio completo con header e property. I messaggi non instradabili tornano indietro su OnAMQPBasicReturn.
AckMessage conferma un delivery tag, RejectMessage lo rifiuta con un requeue facoltativo, e Recover chiede al broker di riconsegnare tutto ciò che è ancora senza conferma.
SelectTransaction mette un canale in modalità transazionale, poi CommitTransaction o RollbackTransaction la chiude. OnAMQPTransactionOk conferma ogni passaggio.
AMQPOptions contiene VirtualHost, MaxChannels, MaxFrameSize e Locale, tutti negoziati con il broker durante l'handshake iniziale.
SetQoS(channel, prefetchSize, prefetchCount, global) limita quanti messaggi senza conferma il broker può inviare a un consumer, così un gestore lento non viene sommerso.
HeartBeat mantiene vive le connessioni inattive attraverso i proxy e oltre i timeout del broker, con OnAMQPHeartBeat sollevato a ogni scambio.
| Guida onlineRiferimento API completo e guida all'utilizzo. | Apri | |
| AMQP 1.0 ClientL'altro AMQP, un protocollo diverso con un componente dedicato. | Apri | |
| Scarica la versione di prova gratuitaEsegui il client contro un container RabbitMQ locale. | Apri | |
| PrezziLicenze Single, Team e Site con codice sorgente completo. | Apri |