RabbitMQ Tutorials in Ballerina #1

Published in

Ballerina Swan Lake Tech Blog

4 min readApr 7, 2023

This article was written using Ballerina Swan Lake Update 8 (2201.8.0)

RabbitMQ is one of the most popular open-source message-queueing solutions, which is commonly known as a message broker. It uses AMQP (Advanced Message Queuing Protocol) to send and receive messages. Ballerina on the other hand is a programming language, which mainly focuses on network applications.

Note: RabbitMQ speaks multiple protocols. All these tutorials use AMQP 0–9–1, which is an open, general-purpose protocol for messaging. The RabbitMQ documentation provides a comprehensive guide to getting started with the broker. There are several clients for RabbitMQ in many different languages. In this tutorial, we’re using the RabbitMQ standard library client provided by Ballerina. The general terms used here are the same as in RabbitMQ documentation.

Setting Up the Environment

Download and install Ballerina.
Download and install RabbitMQ server.
Download and install any code editor.
Basic knowledge of RabbitMQ core concepts.

RabbitMQ Ballerina Client Library

After installing Ballerina, use the below command to pull the latest compatible version of the RabbitMQ Ballerina client library, ballerinax/rabbitmq. For more details on using the library check out the API documentation.

bal pull ballerinax/rabbitmq

You can create two Ballerina projects for the producer and consumer or two simple .bal files as you wish. To learn about Ballerina project structure and getting started, check out Get started with Ballerina.

Tutorial 1: A Simple Hello

Components

Producer: The program that sends the messages.
Queue: Messages flowing through RabbitMQ can only be stored inside a queue. It is essentially a large message buffer. Many producers can send messages that go to one queue, and many consumers can try to receive data from one queue.
Consumer: The program that mostly waits to receive messages.

Note: The producer, consumer, and broker do not have to reside on the same host; in most applications they don’t. An application can be both a producer and consumer.

Sending Messages

First you need to import the ballerinax/rabbitmq module.

import ballerinax/rabbitmq;

The producer will connect to RabbitMQ server, send a single message, then exit. To connect to the RabbitMQ server, we first need to create a rabbitmq:Client which creates the socket connection. The rabbitmq:Client is also the Ballerina interface to provide AMQP Channel-related functionality. The default host and port provided by the module are localhost and 5672 respectively. Here we connect to a RabbitMQ node on the local machine — hence the localhost.

rabbitmq:Client newClient = check new (rabbitmq:DEFAULT_HOST,
                               rabbitmq:DEFAULT_PORT);

To send a message, the queue must be declared beforehand; then we can publish a message to the queue. Declaring a queue is idempotent — it will only be created if it doesn’t exist already.

check newClient->queueDeclare("hello");

The message content is a byte array, check out all the message types supported by the library in the API documentation.

string message = "Hello World!";check newClient->publishMessage({content: message.toBytes(), 
                     routingKey: QUEUE_NAME});

Following is the complete code for the producer.

Receiving Messages

The consumer listens for messages from the RabbitMQ server, so unlike the producer which publishes a single message and exits, we’ll keep the consumer running to listen for incoming messages and print them out.

The most efficient way to receive messages is to set up a subscription using a rabbitmq:Listener and a consumer service. The messages will then be delivered automatically as they arrive rather than having to be explicitly requested. The queue to which the service is listening is configured in the rabbitmq:ServiceConfig annotation of the service or else you can provide it as the name of the service.

listener rabbitmq:Listener rabbitmqListener = new 
                    (rabbitmq:DEFAULT_HOST, rabbitmq:DEFAULT_PORT);@rabbitmq:ServiceConfig {
     queueName: "hello"
}
service on rabbitmqListener {
}

The queue name mentioned for the consumer service is declared internally once the service is started. As we might start the consumer service before the producer is executed, we need to make sure the queue exists before we try to consume messages from it.

The remote function onMessage inside the consumer service will receive the consumed messages and print them out. We’re sending a rabbitmq:BytesMessage to the server, therefore we’re expecting to receive a message of the same type.

remote function onMessage(rabbitmq:BytesMessage message) returns 
        error? {
    string messageContent = check string:fromBytes(message.content);
    log:printInfo("[X] Received message: " + messageContent);
}

Following is the complete code for the consumer.

Compiling and Running

If the producer and consumer are implemented within two different Ballerina projects, you can simply execute the following command inside each of them in separate terminals.

bal run

If the producer and consumer are implemented in two .bal files, then run the following command to execute them.

bal run <path_to_the_file>
eg: 
bal run producer.bal
bal run Desktop/consumer.bal

If the consumer started without any issue, following lines will be printed in the terminal. Let it run in that terminal.

If the producer was able to send the message successfully, the following line will be printed in the terminal.

That’s it! :D Now how easy was that?! Star worthy isn’t it? ;) Thanks!

Checkout other cool libraries provided by Ballerina in Ballerina Central.

Ballerina is an open-source project. We value your feedback and contributions. Please provide feedback, questions, issues related to the Ballerina RabbitMQ package in Ballerina Standard Library GitHub repository.

References:

Versions used in this article:

Ballerina: This article was written using Ballerina Swan Lake Update 8 (2201.8.0)
RabbitMQ client library: ballerinax/rabbitmq:2.10.0