Getting Started to SmallRye Reactive Messaging with AMQP 1.0

This guide demonstrates how your Quarkus application can utilize SmallRye Reactive Messaging to interact with AMQP 1.0.

This technology is considered {extension-status}.

For a full list of possible extension statuses, check our FAQ entry.

If you want to use RabbitMQ, you need to enable AMQP 1.0 in the RabbitMQ broker. Check the connecting to RabbitMQ documentation.

Prerequisites

To complete this guide, you need:

  • less than 15 minutes

  • an IDE

  • JDK 11+ installed with JAVA_HOME configured appropriately

  • Apache Maven 3.8.1+

  • docker and docker-compose

  • GraalVM installed if you want to run in native mode.

Architecture

In this guide, we are going to develop two applications communicating with an AMQP broker. We will use Artemis, but you can use any AMQP 1.0 broker. The first application sends a quote request to an AMQP queue and consumes messages from the quote queue. The second application receives the quote request and sends a quote back.

Architecture

The first application, the producer, will let the user request some quotes over an HTTP endpoint. For each quote request, a random identifier is generated and returned to the user, to put the quote request on pending. At the same time the generated request id is sent over the quote-requests queue.

Producer App UI

The second application, the processor, in turn, will read from the quote-requests queue put a random price to the quote, and send it to a queue named quotes.

Lastly, the producer will read the quotes and send them to the browser using server-sent events. The user will therefore see the quote price updated from pending to the received price in real-time.

Solution

We recommend that you follow the instructions in the next sections and create applications step by step. However, you can go right to the completed example.

Clone the Git repository: git clone https://github.com/quarkusio/quarkus-quickstarts.git, or download an archive.

The solution is located in the amqp-quickstart directory.

Creating the Maven Project

First, we need to create two projects: the producer and the processor.

To create the producer project, in a terminal run:

mvn io.quarkus.platform:quarkus-maven-plugin:2.2.3.Final:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=amqp-quickstart-producer \
    -DnoCode=true \
    -Dextensions=resteasy-reactive-jackson,smallrye-reactive-messaging-amqp

This command creates the project structure and select the two Quarkus extensions we will be using:

  1. RESTEasy Reactive and it’s Jackson support to handle JSON payloads

  2. The Reactive Messaging AMQP connector

To create the processor project, from the same directory, run:

mvn io.quarkus.platform:quarkus-maven-plugin:2.2.3.Final:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=amqp-quickstart-processor \
    -DnoCode=true \
    -Dextensions=smallrye-reactive-messaging-amqp

At that point you should have the following structure:

.
├── amqp-quickstart-processor
│  ├── README.md
│  ├── mvnw
│  ├── mvnw.cmd
│  ├── pom.xml
│  └── src
│     └── main
│        ├── docker
│        ├── java
│        └── resources
│           └── application.properties
└── amqp-quickstart-producer
   ├── README.md
   ├── mvnw
   ├── mvnw.cmd
   ├── pom.xml
   └── src
      └── main
         ├── docker
         ├── java
         └── resources
            └── application.properties

Open the two projects in your favorite IDE.

The Quote object

The Quote class will be used in both producer and processor projects. For the sake of simplicity we will duplicate the class. In both projects, create the src/main/java/org/acme/amqp/model/Quote.java file, with the following content:

package org.acme.amqp.model;

import io.quarkus.runtime.annotations.RegisterForReflection;

@RegisterForReflection
public class Quote {

    public String id;
    public int price;

    /**
    * Default constructor required for Jackson serializer
    */
    public Quote() { }

    public Quote(String id, int price) {
        this.id = id;
        this.price = price;
    }

    @Override
    public String toString() {
        return "Quote{" +
                "id='" + id + '\'' +
                ", price=" + price +
                '}';
    }
}

JSON representation of Quote objects will be used in messages sent to the AMQP queues and also in the server-sent events sent to browser clients.

Quarkus has built-in capabilities to deal with JSON AMQP messages.

@RegisterForReflection

The @RegisterForReflection annotation instructs Quarkus to include the class (including fields and methods) when building the native executable. This will be useful later when we run the applications as native executables inside containers. Without, the native compilation would remove the fields and methods during the dead-code elimination phase.

Sending quote request

Inside the producer project locate the generated src/main/java/org/acme/amqp/producer/QuotesResource.java file, and update the content to be:

package org.acme.amqp.producer;

import java.util.UUID;

import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

import org.acme.amqp.model.Quote;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;

import io.smallrye.mutiny.Multi;

@Path("/quotes")
public class QuotesResource {

    @Channel("quote-requests") Emitter<String> quoteRequestEmitter; (1)

    /**
     * Endpoint to generate a new quote request id and send it to "quote-requests" AMQP queue using the emitter.
     */
    @POST
    @Path("/request")
    @Produces(MediaType.TEXT_PLAIN)
    public String createRequest() {
        UUID uuid = UUID.randomUUID();
        quoteRequestEmitter.send(uuid.toString()); (2)
        return uuid.toString();
    }
}
1 Inject a Reactive Messaging Emitter to send messages to the quote-requests channel.
2 On a post request, generate a random UUID and send it to the AMQP queue using the emitter.

This channel is mapped to an AMQP queue using the configuration we will add to the application.properties file. Open the src/main/resource/application.properties file and add:

# Configure the outgoing `quote-requests` queue
mp.messaging.outgoing.quote-requests.connector=smallrye-amqp

All we need to specify is the smallrye-amqp connector. By default, reactive messaging maps the channel name quote-requests to the AMQP queue name.

Processing quote requests

Now let’s consume the quote request and give out a price. Inside the processor project, locate the src/main/java/org/acme/amqp/processor/QuoteProcessor.java file and add the following:

package org.acme.amqp.processor;

import java.util.Random;

import javax.enterprise.context.ApplicationScoped;

import org.acme.amqp.model.Quote;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import io.smallrye.reactive.messaging.annotations.Blocking;

/**
 * A bean consuming data from the "request" AMQP queue and giving out a random quote.
 * The result is pushed to the "quotes" AMQP queue.
 */
@ApplicationScoped
public class QuoteProcessor {

    private Random random = new Random();

    @Incoming("requests")       (1)
    @Outgoing("quotes")         (2)
    @Blocking                   (3)
    public Quote process(String quoteRequest) throws InterruptedException {
        // simulate some hard working task
        Thread.sleep(200);
        return new Quote(quoteRequest, random.nextInt(100));
    }
}
1 Indicates that the method consumes the items from the requests channel
2 Indicates that the objects returned by the method are sent to the quotes channel
3 Indicates that the processing is blocking and cannot be run on the caller thread.

The process method is called for every AMQP message from the quote-requests queue, and will send a Quote object to the quotes queue.

As with the previous example we need to configure the connectors in the application.properties file. Open the src/main/resources/application.properties file and add:

# Configure the incoming AMQP queue `quote-requests`
mp.messaging.incoming.requests.connector=smallrye-amqp
mp.messaging.incoming.requests.address=quote-requests

# Configure the outgoing AMQP queue `quotes`
mp.messaging.outgoing.quotes.connector=smallrye-amqp

Note that in this case we have one incoming and one outgoing connector configuration, each one distinctly named. The configuration keys are structured as follows:

mp.messaging.[outgoing|incoming].{channel-name}.property=value

The channel-name segment must match the value set in the @Incoming and @Outgoing annotation:

  • quote-requests → AMQP queue from which we read the quote requests

  • quotes → AMQP queue in which we write the quotes

Receiving quotes

Back to our producer project. Let’s modify the QuotesResource to consume quotes, bind it to an HTTP endpoint to send events to clients:

import io.smallrye.mutiny.Multi;
//...

@Channel("quotes") Multi<Quote> quotes;     (1)

/**
 * Endpoint retrieving the "quotes" queue and sending the items to a server sent event.
 */
@GET
@Produces(MediaType.SERVER_SENT_EVENTS) (2)
public Multi<Quote> stream() {
    return quotes; (3)
}
1 Injects the quotes channel using the @Channel qualifier
2 Indicates that the content is sent using Server Sent Events
3 Returns the stream (Reactive Stream)

Again we need to configure the incoming quotes channel inside producer project. Add the following inside application.properties file:

# Configure the outgoing `quote-requests` queue
mp.messaging.outgoing.quote-requests.connector=smallrye-amqp

# Configure the incoming `quotes` queue
mp.messaging.incoming.quotes.connector=smallrye-amqp

The HTML page

Final touch, the HTML page reading the converted prices using SSE.

Create inside the producer project src/main/resources/META-INF/resources/quotes.html file, with the following content:

<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Quotes</title>

    <link rel="stylesheet" type="text/css"
          href="https://cdnjs.cloudflare.com/ajax/libs/patternfly/3.24.0/css/patternfly.min.css">
    <link rel="stylesheet" type="text/css"
          href="https://cdnjs.cloudflare.com/ajax/libs/patternfly/3.24.0/css/patternfly-additions.min.css">
</head>
<body>
<div class="container">
    <div class="card">
        <div class="card-body">
            <h2 class="card-title">Quotes</h2>
            <button class="btn btn-info" id="request-quote">Request Quote</button>
            <div class="quotes"></div>
        </div>
    </div>
</div>
</body>
<script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
<script>
    $("#request-quote").click((event) => {
        fetch("/quotes/request", {method: "POST"})
        .then(res => res.text())
        .then(qid => {
            var row = $(`<h4 class='col-md-12' id='${qid}'>Quote # <i>${qid}</i> | <strong>Pending</strong></h4>`);
            $(".quotes").append(row);
        });
    });
    var source = new EventSource("/quotes");
    source.onmessage = (event) => {
      var json = JSON.parse(event.data);
      $(`#${json.id}`).html(function(index, html) {
        return html.replace("Pending", `\$\xA0${json.price}`);
      });
    };
</script>
</html>

Nothing spectacular here. On each received quote, it updates the page.

Get it running

You just need to run both applications using:

> mvn -f amqp-quickstart-producer quarkus:dev

And, in a separate terminal:

> mvn -f amqp-quickstart-processor quarkus:dev

Quarkus starts a AMQP broker automatically, configures the application and shares the broker instance between different applications. See Dev Services for AMQP for more details.

Open http://localhost:8080/quotes.html in your browser and request some quotes by clicking the button.

Running in JVM or Native mode

When not running in dev or test mode, you will need to start your AMQP broker. You can follow the instructions from the Apache ActiveMQ Artemis website or create a docker-compose.yaml file with the following content:

version: '2'

services:

  artemis:
    image: quay.io/artemiscloud/activemq-artemis-broker:0.1.2
    ports:
      - "8161:8161"
      - "61616:61616"
      - "5672:5672"
    environment:
      AMQ_USER: quarkus
      AMQ_PASSWORD: quarkus
    networks:
      - amqp-quickstart-network

  producer:
    image: quarkus-quickstarts/amqp-quickstart-producer:1.0-${QUARKUS_MODE:-jvm}
    build:
      context: amqp-quickstart-producer
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      AMQP_HOST: artemis
      AMQP_PORT: 5672
    ports:
      - "8080:8080"
    networks:
      - amqp-quickstart-network

  processor:
    image: quarkus-quickstarts/amqp-quickstart-processor:1.0-${QUARKUS_MODE:-jvm}
    build:
      context: amqp-quickstart-processor
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      AMQP_HOST: artemis
      AMQP_PORT: 5672
    networks:
      - amqp-quickstart-network

networks:
  amqp-quickstart-network:
    name: amqp-quickstart

Note how the AMQP broker location is configured. The amqp.host and amqp.port (AMQP_HOST and AMQP_PORT environment variables) properties configure location.

First, make sure you stopped the applications, and build both applications in JVM mode with:

> mvn -f amqp-quickstart-producer clean package
> mvn -f amqp-quickstart-processor clean package

Once packaged, run docker compose up --build. The UI is exposed on http://localhost:8080/quotes.html

To run your applications as native, first we need to build the native executables:

> mvn -f amqp-quickstart-producer package -Pnative  -Dquarkus.native.container-build=true
> mvn -f amqp-quickstart-processor package -Pnative -Dquarkus.native.container-build=true

The -Dquarkus.native.container-build=true instructs Quarkus to build Linux 64bits native executables, who can run inside containers. Then, run the system using:

> export QUARKUS_MODE=native
> docker compose up --build

As before, the UI is exposed on http://localhost:8080/quotes.html

Going further

This guide has shown how you can interact with AMQP 1.0 using Quarkus. It utilizes SmallRye Reactive Messaging to build data streaming applications.

If you did the Kafka quickstart, you have realized that it’s the same code. The only difference is the connector configuration and the JSON mapping.