To demonstrate cross-platform message exchange, code is provided in Delphi and in Java.
- 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.
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.
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;
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;
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.
This folder contains the Delphi project EmitLogDirect and the Java project ReceiveLogDirect
This folder contains the Java project EmitLogDirect and the Delphi project ReceiveLogDirect
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.