Habari Client for RabbitMQ: Direct Exchange Routing

New example projects included with Habari Client libraries 6.1 are based on the RabbitMQ routing tutorial code.

To demonstrate cross-platform message exchange, code is provided in Delphi and in Java.

Delphi projects

• EmitLogDirect – sends a log message with severity (info, warning or error) to the exchange. Severity and message text can be passed as command line arguments.
• ReceiveLogsDirect – receive log messages. The severity can be passed as command line arguments.

Example

To start receiving info, warning and error messages launch the ReceiveLogsDirect program and pass the severity level names

>ReceiveLogsDirect info warning error 
[*] Waiting for logs. To exit press CTRL+C

Tweaking ITopic destination objects to access RabbitMQ exchanges

The Delphi code shows how a client can work with AMQP exchange destinations using a STOMP connection. In the RabbitMQ STOMP plugin, Queues and Topics are the basic destination types. The RabbitMQ STOMP plugin however also offers a way to communicate with AMQP exchange destinations. This requires a special syntax for the destination name. It is neccessary to specify the destination name as /exchange/<name>[/<routing-key>] to send messages, or /exchange/<name>[/<pattern>] to subscribe. For details, see the docs for the RabbitMQ STOMP client. With the Habari Client library, a Topic destination object may be tweaked to access an exchange.

Code excerpts

EmitLogDirect

  Severity := GetSeverity;
  Message := GetMessage;

  ExchangeAndRouting := '/exchange/' + EXCHANGE_NAME + '/' + Severity;

  Factory := TBTConnectionFactory.Create;
  Context := Factory.CreateContext;
  try
    Context.Start;

    Destination := Context.CreateTopic(ExchangeAndRouting);
    Producer := Context.CreateProducer;
    Producer.Send(Destination, GetMessage);

    WriteLn(' [x] Sent ''' + Severity + ''':''' + Message);

  finally
    Context.Close;
  end;

ReceiveLogsDirect

The Delphi project subscribes to the routing keys which are specified as command line args, by creating destinations in a loop over ParamStr().

Note that the connection URL must include the message.subscription_check=false parameter to allow multiple subscriptions for one consumer.

  Factory := TBTConnectionFactory.Create('stomp://localhost?message.subscription_check=false');
  Context := Factory.CreateContext;
  try
    Context.Start;

    WriteLn(' [*] Waiting for messages. To exit press CTRL+C');

    for I := 1 to ParamCount do begin
      Severity := ParamStr(I);
      ExchangeAndRouting := '/exchange/' + EXCHANGE_NAME + '/' + Severity;
      Destination := Context.CreateTopic(ExchangeAndRouting);
      Consumer := Context.CreateConsumer(Destination);
    end;

    while True do begin
      TextMessage := Consumer.Receive(2500) as ITextMessage;
      if Assigned(TextMessage) then
      begin
        WriteLn(' [x] Received ''(severity)'':''' + TextMessage.Text + '''');
      end;
    end;

  finally
    Context.Close;
  end;

 Location

The example code is in the demo-broker directory. There are two subfolders. Both contain a Delphi and a Java project, one for sending and one for receiving messages.

routing-delphi-java

This folder contains the Delphi project EmitLogDirect and the Java project ReceiveLogDirect

routing-java-delphi

This folder contains the Java project EmitLogDirect and the Delphi project ReceiveLogDirect

Availability

The code will be included in the next development snapshot of Habari Client for RabbitMQ.

Production [Non-]Suitability Disclaimer

Please keep in mind that this and other tutorials are, well, tutorials. They demonstrate one new concept at a time and may intentionally oversimplify some things and leave out others. For example topics such as connection management, error handling, connection recovery, concurrency and metric collection are largely omitted for the sake of brevity. Such simplified code should not be considered production ready.

About Habari Client libraries

Habari Client libraries enable Object Pascal applications to take advantage of message broker / message queue technology – which is distributed, loosely coupled, reliable and asynchronous – to build integrated systems, using peer-to-peer and publish-subscribe communication models.

Advertisement