STOMP (RabbitMQ)

Overview

This is the Client Protocol STOMP Component for RabbitMQ Broker.

At a glance

Features

• Native Delphi implementation with full ANSI/Unicode support.

Technical specification

Main properties

The principal published / public properties used to configure and drive the component. Consult the online help for the full list.

Main methods

The principal public methods exposed by the component.

Public events

The component exposes the following published events; consult the online help for full event-handler signatures.

Quick Start

Drop the component on a form, configure the properties below and activate it. The snippet that follows shows the typical Connect to the RabbitMQ Web-STOMP plugin configuration.

Delphi (VCL / FireMonkey)

oClient := TsgcWebSocketClient.Create(nil);
oClient.Host := 'rabbit.example.com';
oClient.Port := 15674;
oClient.Options.Parameters := '/ws';

oRabbitMQ := TsgcWSPClient_STOMP_RabbitMQ.Create(nil);
oRabbitMQ.Client := oClient;
oRabbitMQ.Authentication.Enabled := True;
oRabbitMQ.Authentication.UserName := 'guest';
oRabbitMQ.Authentication.Password := 'guest';
oRabbitMQ.Options.VirtualHost := '/';
// HeartBeat values are milliseconds
oRabbitMQ.HeartBeat.Enabled := True;
oRabbitMQ.HeartBeat.Outgoing := 10000;
oRabbitMQ.HeartBeat.Incoming := 10000;
oRabbitMQ.OnRabbitMQConnected := OnRabbitMQConnectedEvent;
oRabbitMQ.OnRabbitMQMessage := OnRabbitMQMessageEvent;

oClient.Active := True;

C++ Builder

oClient = new TsgcWebSocketClient(this);
oClient->Host = "rabbit.example.com";
oClient->Port = 15674;
oClient->Options->Parameters = "/ws";

oRabbitMQ = new TsgcWSPClient_STOMP_RabbitMQ(this);
oRabbitMQ->Client = oClient;
oRabbitMQ->Authentication->Enabled = true;
oRabbitMQ->Authentication->UserName = "guest";
oRabbitMQ->Authentication->Password = "guest";
oRabbitMQ->Options->VirtualHost = "/";
oRabbitMQ->HeartBeat->Enabled = true;
oRabbitMQ->HeartBeat->Outgoing = 10000;
oRabbitMQ->HeartBeat->Incoming = 10000;
oRabbitMQ->OnRabbitMQConnected = OnRabbitMQConnectedEvent;
oRabbitMQ->OnRabbitMQMessage = OnRabbitMQMessageEvent;

oClient->Active = true;

.NET (C#)

// the .NET client exposes the flat OnSTOMP* events for every STOMP broker
oClient = new TsgcWebSocketClient();
oClient.Host = "rabbit.example.com";
oClient.Port = 15674;
oClient.Options.Parameters = "/ws";

oRabbitMQ = new TsgcWSPClient_STOMP_RabbitMQ();
oRabbitMQ.Client = oClient;
oRabbitMQ.Authentication.Enabled = true;
oRabbitMQ.Authentication.UserName = "guest";
oRabbitMQ.Authentication.Password = "guest";
oRabbitMQ.Options.VirtualHost = "/";
oRabbitMQ.HeartBeat.Enabled = true;
oRabbitMQ.HeartBeat.Outgoing = 10000;
oRabbitMQ.HeartBeat.Incoming = 10000;
oRabbitMQ.OnSTOMPConnected += OnSTOMPConnectedEvent;
oRabbitMQ.OnSTOMPMessage += OnSTOMPMessageEvent;

oClient.Active = true;

Common scenarios

Each scenario shows the configuration and method calls needed to drive the component through a specific real-world flow. Every identifier below is taken from the component declaration shipped with the library.

1 · Topics and gateway queues

SubscribeTopic and PublishTopic use the /topic/name destination prefix, SubscribeQueue and PublishQueue use /queue/name, the queue being declared by the STOMP gateway. Subscribe once the CONNECTED frame has arrived, that is, inside the OnRabbitMQConnected handler.

procedure TForm1.OnRabbitMQConnectedEvent(Connection: TsgcWSConnection;
  Headers: TsgcWSRabbitMQSTOMPHeadersConnected);
begin
  oRabbitMQ.SubscribeTopic('quotes');
  oRabbitMQ.SubscribeQueue('orders');

  oRabbitMQ.PublishTopic('quotes', 'AAPL 195.30');
  oRabbitMQ.PublishQueue('orders', '{"orderId": 12345}');
end;

procedure TForm1.OnRabbitMQMessageEvent(Connection: TsgcWSConnection;
  MessageText: String; Headers: TsgcWSRabbitMQSTOMPHeadersMessage;
  Subscription: TsgcWSBrokerSTOMPSubscriptionItem);
begin
  DoLog(Headers.Destination + ': ' + MessageText);
end;

void __fastcall TForm1::OnRabbitMQConnectedEvent(TsgcWSConnection *Connection,
  TsgcWSRabbitMQSTOMPHeadersConnected *Headers)
{
  oRabbitMQ->SubscribeTopic("quotes");
  oRabbitMQ->SubscribeQueue("orders");

  oRabbitMQ->PublishTopic("quotes", "AAPL 195.30");
  oRabbitMQ->PublishQueue("orders", "{\"orderId\": 12345}");
}

void __fastcall TForm1::OnRabbitMQMessageEvent(TsgcWSConnection *Connection,
  String MessageText, TsgcWSRabbitMQSTOMPHeadersMessage *Headers,
  TsgcWSBrokerSTOMPSubscriptionItem *Subscription)
{
  DoLog(Headers->Destination + ": " + MessageText);
}

void OnSTOMPConnectedEvent(TsgcWSConnection Connection, string Version,
  string Server, string Session, string HeartBeat, string RawHeaders)
{
  oRabbitMQ.SubscribeTopic("quotes");
  oRabbitMQ.SubscribeQueue("orders");

  oRabbitMQ.PublishTopic("quotes", "AAPL 195.30");
  oRabbitMQ.PublishQueue("orders", "{\"orderId\": 12345}");
}

void OnSTOMPMessageEvent(TsgcWSConnection Connection, string MessageText,
  string Destination, string MessageId, string Subscription,
  string ACK, string ContentType, string RawHeaders)
{
  DoLog(Destination + ": " + MessageText);
}

2 · Queues declared outside STOMP, and exchanges

SubscribeQueueOutside and PublishQueueOutside address a queue that was declared through AMQP or the management UI, using the /amq/queue/name prefix. SubscribeExchange binds a pattern on an exchange, and PublishExchange sends to a routing key. Queue arguments such as message TTL or a dead-letter exchange are set through the Queue property.

oRabbitMQ.Queue.MessageTTL := 60000;
oRabbitMQ.Queue.MaxLength := 1000;
oRabbitMQ.Queue.DeadLetterExchange := 'dlx';

// queue declared outside the STOMP gateway: /amq/queue/invoices
oRabbitMQ.SubscribeQueueOutside('invoices');
oRabbitMQ.PublishQueueOutside('invoices', 'INV-2026-001');

// exchange binding: SUBSCRIBE to a pattern, SEND to a routing key
oRabbitMQ.SubscribeExchange('logs', 'app.*');
oRabbitMQ.PublishExchange('logs', 'app.error', 'disk full');

oRabbitMQ->Queue->MessageTTL = 60000;
oRabbitMQ->Queue->MaxLength = 1000;
oRabbitMQ->Queue->DeadLetterExchange = "dlx";

oRabbitMQ->SubscribeQueueOutside("invoices");
oRabbitMQ->PublishQueueOutside("invoices", "INV-2026-001");

oRabbitMQ->SubscribeExchange("logs", "app.*");
oRabbitMQ->PublishExchange("logs", "app.error", "disk full");

oRabbitMQ.Queue.MessageTTL = 60000;
oRabbitMQ.Queue.MaxLength = 1000;
oRabbitMQ.Queue.DeadLetterExchange = "dlx";

oRabbitMQ.SubscribeQueueOutside("invoices");
oRabbitMQ.PublishQueueOutside("invoices", "INV-2026-001");

oRabbitMQ.SubscribeExchange("logs", "app.*");
oRabbitMQ.PublishExchange("logs", "app.error", "disk full");

3 · Acknowledgements and transactions

Broker_Options.Acknowledgments decides who sends the ACK frame. Leave it at ackAutomatic and the component acknowledges every message for you, switch it to ackManual and you call ACK or ACKEx yourself from the message handler. BeginTransaction, CommitTransaction and AbortTransaction group SEND and ACK frames into one atomic unit.

// take over the acknowledgement, and subscribe requesting client ACK
oRabbitMQ.Broker_Options.Acknowledgments := ackManual;
oRabbitMQ.SubscribeQueue('orders', True, False, False, ackIndividual);

procedure TForm1.OnRabbitMQMessageEvent(Connection: TsgcWSConnection;
  MessageText: String; Headers: TsgcWSRabbitMQSTOMPHeadersMessage;
  Subscription: TsgcWSBrokerSTOMPSubscriptionItem);
begin
  // STOMP 1.2 uses the ack header, 1.1 and 1.0 the message-id header
  if Headers.ACK <> '' then
    oRabbitMQ.ACKEx(Headers.ACK, Subscription.Id)
  else
    oRabbitMQ.ACKEx(Headers.MessageId, Subscription.Id);
end;

// atomic group of SEND frames
oRabbitMQ.BeginTransaction('tx-1');
oRabbitMQ.PublishQueue('orders', 'first', 'text/plain', 'tx-1');
oRabbitMQ.PublishQueue('orders', 'second', 'text/plain', 'tx-1');
oRabbitMQ.CommitTransaction('tx-1');

oRabbitMQ->Broker_Options->Acknowledgments = ackManual;
oRabbitMQ->SubscribeQueue("orders", true, false, false, ackIndividual);

void __fastcall TForm1::OnRabbitMQMessageEvent(TsgcWSConnection *Connection,
  String MessageText, TsgcWSRabbitMQSTOMPHeadersMessage *Headers,
  TsgcWSBrokerSTOMPSubscriptionItem *Subscription)
{
  if (Headers->ACK != "")
    oRabbitMQ->ACKEx(Headers->ACK, Subscription->Id);
  else
    oRabbitMQ->ACKEx(Headers->MessageId, Subscription->Id);
}

oRabbitMQ->BeginTransaction("tx-1");
oRabbitMQ->PublishQueue("orders", "first", "text/plain", "tx-1");
oRabbitMQ->PublishQueue("orders", "second", "text/plain", "tx-1");
oRabbitMQ->CommitTransaction("tx-1");

// 0 = auto, 1 = client (multiple), 2 = client-individual
oRabbitMQ.BrokerOptions.Acknowledgments = TsgcWSBrokerACK.Manual;
oRabbitMQ.SubscribeQueue("orders", true, false, false, 2);

void OnSTOMPMessageEvent(TsgcWSConnection Connection, string MessageText,
  string Destination, string MessageId, string Subscription,
  string ACK, string ContentType, string RawHeaders)
{
  oRabbitMQ.ACK(ACK != "" ? ACK : MessageId);
}

Sources used to build this document

Every external claim links back to a primary source. The online-help references decode the canonical deep-link the company maintains for this component.