TsgcWSPClient_AMQP habla AMQP 0.9.1, el protocolo alrededor del cual se diseñó RabbitMQ. Expone el modelo tal y como es: abres un canal, declaras un exchange y una cola, los enlazas con una routing key y después publicas y consumes. No hay nada escondido tras una fachada simplificada, así que el broker se comporta como dice su propia documentación.
Los canales se nombran con la cadena que tú elijas, así que 'ch1' identifica el mismo canal en todas las llamadas, en lugar de un id numérico del que tengas que llevar la cuenta.
TsgcWSPClient_AMQP
AMQP 0-9-1
TCP (5672) o TLS (5671)
Delphi, C++ Builder
Transporte: TCP plano y TLS. sgcMQ se conecta por TCP plano y TLS. Si necesitas AMQP sobre WebSocket, eso requiere un paquete sgcWebSockets, que es el que aporta el cliente WebSocket.
Haz las declaraciones dentro de OnAMQPConnect, una vez terminado el handshake. Las entregas llegan después a 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}");Casi todos los métodos tienen un gemelo bloqueante ...Ex que espera la respuesta del broker y te la devuelve, que es justo lo que quieres en una rutina de configuración lineal.
OpenChannel y CloseChannel multiplexan varias conversaciones lógicas sobre una única conexión TCP. EnableChannel y DisableChannel aplican el control de flujo del canal.
DeclareExchange(channel, name, type) crea un exchange de tipo direct, fanout, topic o headers. DeleteExchange lo elimina.
DeclareQueue, BindQueue, UnBindQueue, PurgeQueue y DeleteQueue cubren todo el ciclo de vida de una cola, cada uno con una variante ...Ex que devuelve la respuesta del broker.
Consume(channel, queue) inicia una suscripción y CancelConsume la termina. Las entregas llegan a OnAMQPBasicDeliver, con OnAMQPBasicGetOk y OnAMQPBasicGetEmpty para las recogidas de un solo mensaje.
PublishMessage(channel, exchange, routingKey, body) tiene sobrecargas para texto, streams y un objeto de mensaje completo con cabeceras y propiedades. Los mensajes no enrutables vuelven por OnAMQPBasicReturn.
AckMessage confirma un delivery tag, RejectMessage lo rechaza con reencolado opcional, y Recover pide al broker que reentregue todo lo que sigue sin confirmar.
SelectTransaction pone un canal en modo transaccional, y después CommitTransaction o RollbackTransaction lo cierra. OnAMQPTransactionOk confirma cada paso.
AMQPOptions reúne VirtualHost, MaxChannels, MaxFrameSize y Locale, todos ellos negociados con el broker durante el handshake inicial.
SetQoS(channel, prefetchSize, prefetchCount, global) limita cuántos mensajes sin confirmar puede empujar el broker a un consumidor, de modo que un manejador lento no se vea desbordado.
HeartBeat mantiene vivas las conexiones inactivas a través de proxies y más allá de los tiempos de espera del broker, con OnAMQPHeartBeat en cada intercambio.
| Ayuda en líneaReferencia completa de la API y guía de uso. | Abrir | |
| AMQP 1.0 ClientEl otro AMQP, un protocolo distinto con su propio componente. | Abrir | |
| Descargar prueba gratuitaPrueba el cliente contra un contenedor local de RabbitMQ. | Abrir | |
| PreciosLicencias Single, Team y Site con código fuente completo. | Abrir |