Developing a simple reactive messaging application with Vert.x

In this tutorial, we’ll create a simple reactive application which will receive messages from IBM MQ. We will be running this application as a standalone Java application.

Reactive messaging is an asynchronous communication strategy designed to manage resources as efficiently as possible. Instead of actively checking for any requests, reactive programs execute a task only after a message is available on a queue. This reactive paradigm allows the application to be flexible in its responses and be easily scalable.

This application will use the Vert.x framework to receive messages from a topic on which we’ll publish the messages using the MQ console. Vert.x will act as a bridge between our application and the MQ queue manager, allowing both technologies to interact with each other using AMQP.

Vert.x is a flexible and lightweight framework for reactive applications. The Vert.x framework has an AMQP client that uses publish/subscribe messaging methods. When we create a “receiver” application, Vert.x creates a subscription that waits and listens for incoming messages to display. We will use the framework's AMQP client to interact with the MQ queue manager.

Prerequisites

To complete this tutorial, you’ll need to install or set up:

• Git, Maven 3.8.3, and Java JDK 11
• Docker or Podman
• An AMQP channel enabled instance of IBM MQ. Follow the steps in this tutorial

Steps

Step 1. Clone the repository

Make sure you are in a directory where you want to clone the mq-dev-patterns repository, and then clone it by running this command:

git clone https://github.com/ibm-messaging/mq-dev-patterns.git

After cloning the repo, cd into the project directory:

cd reactive/amqp-vertx

Step 2. Set up the environment variables and dependencies

Before understanding how our application works, let’s talk about the environment variables we will need.

In order for our reactive app to securely communicate with MQ, we need to set the host, port, username, and password environment variables to configure our Vert.x AMQP client.

The password you need to specify on this environment variable is the one you used when you set up your MQ queue manager.

Copy and paste the following in your terminal to set the environment variables:

On Mac:

export HOST_NAME=localhost
export PORT=5672
export MQ_USER=app
export PASSWORD=passw0rd

On Windows:

set HOST_NAME=localhost
set PORT=5672
set MQ_USER=app
set PASSWORD=passw0rd

If you are a macOS user, we need to modify the default pom.xml file to add an additional dependency.

We need to add the Vert.x AMQP Client as a dependency in our Java application so that we can use the Vert.x framework to connect to the MQ queue manager.

If you are a macOS user, you need to add the following lines between the lines and . If you are a Windows or Linux user, you do not need to do this step.

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-resolver-dns-native-macos</artifactId>
    <version>4.1.70.Final</version>
    <classifier>osx-x86_64</classifier>
</dependency>

Step 3. Create the Vert.x receiver application

There are three Java files that make up our receiver application. These files are AmqpClientAndConnection.java, Receiver.java, and Runner.java. Runner.java is the main file, and it creates the objects we are going to need.

We are going to use our AmqpClientAndConnection object to initialize the process of creating a connection. Initializing the AmqpClientAndConnection object gets the program to check if the environment variables were provided correctly. If they are not provided correctly, the program exits.

public AmqpClientAndConnection() {
  if (envSettingsProvided()) {
    this.host = System.getenv("HOST_NAME");
    this.port = Integer.valueOf(System.getenv("PORT"));
    this.username = System.getenv("MQ_USER");
    this.password = System.getenv("PASSWORD");

    // Initialise client and connection.
    client = null;
    connection = null;
  } else {
    System.out.println("The environment settings were not provided correctly.");
    System.exit(0);
  }
}

If the environment variable check in AmqpClientAndConnection.java succeeds, our program then checks to see if there is an existing client connection.

if (client == null) {
    System.out.println("There is no existing client. Trying to create one. ");
    createClient();
}

If there is no existing client, we will create our client in the same file using the following lines:

public void createClient() {
    AmqpClientOptions options = new AmqpClientOptions()
    .setHost(host)
    .setPort(port)
    .setUsername(username)
    .setPassword(password);
    // Create a client using its own internal Vert.x instance.
    this.client = AmqpClient.create(options);
    System.out.println("Created a client.");
}

In production environments, you will want to set SSL/TLS protocols most likely as environment variables to provide as an option in the app. For more information about the various AMQP Client options, see the Vert.x docs.

We then create our connection using this client with the following code:

client.connect(ar -> {
    if (ar.failed()) {
        System.out.println("Unable to connect to the broker");
    } else {
        this.connection = ar.result();
        System.out.println("Created a connection.");
    }
});

After the connection has been established, we will use it to create a receiver. This receiver will be created in the Receiver.java file in the same directory. This receiver object will display messages sent to an address named “myreceiver”.

public void createReceiver(AmqpClientAndConnection clientAndConnection){
    clientAndConnection.getConnection().createReceiver("myreceiver",
        done -> {
            if (done.failed()) {
                System.out.println("Unable to create receiver");
                clientAndConnection.closeConnection();
            } else {
                this.receiver = done.result();
                receiver.handler(msg -> {
                    System.out.println("Received the message: " + msg.bodyAsString());
                });
            }
        }
    );
    // Wait max of 15 seconds for the connection to be made.
    waitForReceiverToEstablish();
}

If the receiver cannot be created for any reason, the program proceeds to close the connection.

clientAndConnection.closeConnection();

The following snippet displays every message the receiver gets on the terminal:

receiver.handler(msg -> {
    System.out.println("Received the message: " + msg.bodyAsString());
});

Finally, we will include a catch for Ctrl + C from the user to close the existing connection and exit the program cleanly:

Runtime.getRuntime().addShutdownHook(new Thread()
{
    @Override
    public void run()
    {
         clientAndConnection.closeConnection();
    }
});

Step 4. Compile and run the reactive application

Now, compile the Java app using this command:

mvn clean package dependency:copy-dependencies

Before running the app, make sure your local MQ queue manager is up and running. You can check this with the command:

docker ps

or

podman ps

We can now run the program using this command:

java -cp "target/receiver-example-1.jar:target/dependency/*" vertxtutorial.Runner

If you see the following messages on your terminal after the connection initialization process has begun, don’t worry. They indicate DNS and network related latency, but the connection should be established after a few of these warnings:

After running the Receiver app, navigate to the MQ console by typing in the URL https://localhost:9443 and log in.

After logging in, navigate to the Manage QM1 section and check that you can see myreceiver under the Subscriptions tab.

Click the subscription myreceiver.

Click the Create button to create a message.

Publish a test message to the receiver.

After putting the message on “myreceiver”, look in the terminal where you ran the Receiver app. You should now see the message you created.

Congratulations! You have now successfully created a basic receiver application that displays the message coming from an MQ queue manager using AMQP protocol.

Summary and next steps

In this tutorial, we created a reactive application to receive messages from our MQ queue manager using Vert.x. We created an AMQP client to create a connection to our MQ queue manager. Inside this application, we also created a receiver object using this connection to listen for any messages to display.

You can now explore more about what Vert.x offers to create a reactive application that communicates with an MQ queue manager.