STOMP (ActiveMQ)

Overview

This is the Client Protocol STOMP Component for ActiveMQ 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 Apache ActiveMQ configuration.

Delphi (VCL / FireMonkey)

oClient := TsgcWebSocketClient.Create(nil);
oClient.Host := 'broker.example.com';
oClient.Port := 61614;

oActiveMQ := TsgcWSPClient_STOMP_ActiveMQ.Create(nil);
oActiveMQ.Client := oClient;
oActiveMQ.Authentication.Enabled := True;
oActiveMQ.Authentication.UserName := 'admin';
oActiveMQ.Authentication.Password := 'admin';
// ClientId is required for durable topic subscriptions
oActiveMQ.ActiveMQ_Options.ClientId := 'delphi-client-1';
// HeartBeat values are milliseconds
oActiveMQ.HeartBeat.Enabled := True;
oActiveMQ.HeartBeat.Outgoing := 10000;
oActiveMQ.HeartBeat.Incoming := 10000;
oActiveMQ.OnActiveMQConnected := OnActiveMQConnectedEvent;
oActiveMQ.OnActiveMQMessage := OnActiveMQMessageEvent;

oClient.Active := True;

C++ Builder

oClient = new TsgcWebSocketClient(this);
oClient->Host = "broker.example.com";
oClient->Port = 61614;

oActiveMQ = new TsgcWSPClient_STOMP_ActiveMQ(this);
oActiveMQ->Client = oClient;
oActiveMQ->Authentication->Enabled = true;
oActiveMQ->Authentication->UserName = "admin";
oActiveMQ->Authentication->Password = "admin";
oActiveMQ->ActiveMQ_Options->ClientId = "cbuilder-client-1";
oActiveMQ->HeartBeat->Enabled = true;
oActiveMQ->HeartBeat->Outgoing = 10000;
oActiveMQ->HeartBeat->Incoming = 10000;
oActiveMQ->OnActiveMQConnected = OnActiveMQConnectedEvent;
oActiveMQ->OnActiveMQMessage = OnActiveMQMessageEvent;

oClient->Active = true;

.NET (C#)

// the .NET client exposes the flat OnSTOMP* events for every STOMP broker,
// and names the broker options ActiveMQOptions without the underscore
oClient = new TsgcWebSocketClient();
oClient.Host = "broker.example.com";
oClient.Port = 61614;

oActiveMQ = new TsgcWSPClient_STOMP_ActiveMQ();
oActiveMQ.Client = oClient;
oActiveMQ.Authentication.Enabled = true;
oActiveMQ.Authentication.UserName = "admin";
oActiveMQ.Authentication.Password = "admin";
oActiveMQ.ActiveMQOptions.ClientId = "net-client-1";
oActiveMQ.HeartBeat.Enabled = true;
oActiveMQ.HeartBeat.Outgoing = 10000;
oActiveMQ.HeartBeat.Incoming = 10000;
oActiveMQ.OnSTOMPConnected += OnSTOMPConnectedEvent;
oActiveMQ.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 queues

SubscribeTopic and PublishTopic use the /topic/name destination prefix, SubscribeQueue and PublishQueue use /queue/name. Subscribe once the CONNECTED frame has arrived, that is, inside the OnActiveMQConnected handler. Note that the ActiveMQ subscribe methods take aDurable and aExclusive, they have no aAutoDelete parameter.

procedure TForm1.OnActiveMQConnectedEvent(Connection: TsgcWSConnection;
  Headers: TsgcWSActiveMQSTOMPHeadersConnected);
begin
  oActiveMQ.SubscribeTopic('quotes');
  oActiveMQ.SubscribeQueue('orders');

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

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

void __fastcall TForm1::OnActiveMQConnectedEvent(TsgcWSConnection *Connection,
  TsgcWSActiveMQSTOMPHeadersConnected *Headers)
{
  oActiveMQ->SubscribeTopic("quotes");
  oActiveMQ->SubscribeQueue("orders");

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

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

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

  oActiveMQ.PublishTopic("quotes", "AAPL 195.30");
  oActiveMQ.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 · ActiveMQ subscription headers and JMS message headers

The Queue property carries the ActiveMQ subscription extensions, a JMS selector, a prefetch size, no-local and dispatch-async. On the sending side a TsgcWSActiveMQSTOMP_Message_Options instance adds the JMS headers ActiveMQ understands, correlation id, expiration, message groups, persistence, priority and reply-to.

// subscription extensions
oActiveMQ.Queue.Selector := 'JMSPriority > 5';
oActiveMQ.Queue.PrefetchSize := 100;
oActiveMQ.Queue.NoLocal := True;
oActiveMQ.Queue.DispatchAsync := True;
oActiveMQ.SubscribeQueue('orders');

// JMS headers on a published message
oOptions := TsgcWSActiveMQSTOMP_Message_Options.Create;
try
  oOptions.CorrelationId := 'req-42';
  oOptions.Persistent := True;
  oOptions.Priority := 9;
  oOptions.ReplyTo := '/queue/replies';
  oOptions.JMSXGroupID := 'group-a';
  oActiveMQ.PublishQueue('orders', 'payload', 'text/plain', '', oOptions);
finally
  oOptions.Free;
end;

oActiveMQ->Queue->Selector = "JMSPriority > 5";
oActiveMQ->Queue->PrefetchSize = 100;
oActiveMQ->Queue->NoLocal = true;
oActiveMQ->Queue->DispatchAsync = true;
oActiveMQ->SubscribeQueue("orders");

oOptions = new TsgcWSActiveMQSTOMP_Message_Options();
try {
  oOptions->CorrelationId = "req-42";
  oOptions->Persistent = true;
  oOptions->Priority = 9;
  oOptions->ReplyTo = "/queue/replies";
  oOptions->JMSXGroupID = "group-a";
  oActiveMQ->PublishQueue("orders", "payload", "text/plain", "", oOptions);
} __finally {
  delete oOptions;
}

// the .NET wrapper exposes the subscription extensions through Queue,
// PublishQueue takes destination, body and content type
oActiveMQ.Queue.Selector = "JMSPriority > 5";
oActiveMQ.Queue.PrefetchSize = 100;
oActiveMQ.Queue.NoLocal = true;
oActiveMQ.Queue.DispatchAsync = true;
oActiveMQ.SubscribeQueue("orders");

oActiveMQ.PublishQueue("orders", "payload", "text/plain");

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
oActiveMQ.Broker_Options.Acknowledgments := ackManual;
oActiveMQ.SubscribeQueue('orders', True, False, ackIndividual);

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

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

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

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

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

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

void OnSTOMPMessageEvent(TsgcWSConnection Connection, string MessageText,
  string Destination, string MessageId, string Subscription,
  string ACK, string ContentType, string RawHeaders)
{
  oActiveMQ.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.