Quarkus - Asynchronous messages between beans

• point-to-point - send the message, one consumer receives it. If several consumers listen to the address, a round robin is applied;

• publish/subscribe - publish a message, all the consumers listening to the address are receiving the message;

• request/reply - send the message and expect a response. The receiver can respond to the message in an asynchronous-fashion

All these delivery mechanism are non-blocking, and are providing one of the fundamental brick to build reactive applications.

This mechanism uses the Vert.x EventBus, so you need to enable the extension to use this feature.If you are creating a new project, set the extensions parameter are follows:

If you have an already created project, the vertx extension can be added to an existing Quarkus project withthe add-extension command:

./mvnw quarkus:add-extension -Dextensions="vertx"

Otherwise, you can manually add this to the dependencies section of your pom.xml file:

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-vertx</artifactId>
</dependency>

To consume events, use the io.quarkus.vertx.ConsumeEvent annotation:

package org.acme.vertx;
import io.quarkus.vertx.ConsumeEvent;
import javax.enterprise.context.ApplicationScoped;
@ApplicationScoped
public class GreetingService {
    @ConsumeEvent                           (1)
    public String consume(String name) {    (2)
        return name.toUpperCase();
    }
}

The @ConsumeEvent annotation can be configured to set the address:

@ConsumeEvent("greeting")               (1)
public String consume(String name) {
    return name.toUpperCase();
}

The return value of a method annotated with @ConsumeEvent is used as response to the incoming message.For instance, in the following snippet, the returned String is the response.

You can also return a CompletionStage<T> to handle asynchronous reply:

@ConsumeEvent("greeting")
public CompletionStage<String> consume2(String name) {
    return CompletableFuture.supplyAsync(name::toUpperCase, executor);
}

@ConsumeEvent("greeting")
public void consume(String event) {
    // Do something with the event

As said above, this mechanism is based on the Vert.x event bus. So, you can also use Message directly:

@ConsumeEvent("greeting")
public void consume(Message<String> msg) {
    System.out.println(msg.address());
    System.out.println(msg.body());
}

Ok, we have seen how to receive messages, let’s now switch to the other side: the sender.Sending and publishing messages use the Vert.x event bus:

package org.acme;
import io.vertx.axle.core.eventbus.EventBus;
import io.vertx.axle.core.eventbus.Message;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import java.util.concurrent.CompletionStage;
@Path("/async")
public class EventResource {
  @Inject
  EventBus bus;                                                        (1)
  @GET
  @Path("/{name}")
  public CompletionStage<String> hello(String name) {
      return bus.<String>send("greeting", name)                             (2)
              .thenApply(Message::body);
  }
}

The EventBus object provides methods to:

• send a message to a specific address - one single consumer receives the message.

• publish a message to a specific address - all consumers receive the messages.

• send a message and expect reply

// Case 1
bus.send("address", "hello");
// Case 2
bus.publish("address", "hello");
// Case 3
bus.send("address", "hello, how are you?").thenAccept(message -> {
    // reponse
});

Let’s revisit a greeting HTTP endpoint and use asynchronous message passing to delegate the call to a separated bean.It uses the request/reply dispatching mechanism.Instead of implementing the business logic inside the JAX-RS endpoint, we are sending a message.This message is consumed by another bean and the response is sent using the reply mechanism.

First create a new project using:

You can already start the application in dev mode using .

Then, creates a new JAX-RS resource with the following content:

src/main/java/org/acme/vertx/EventResource.java

package org.acme.vertx;
import io.vertx.axle.core.eventbus.Message;
import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import java.util.concurrent.CompletionStage;
@Path("/hello")
public class EventResource {
    @Inject EventBus bus;
    @GET
    @Path("/async/{name}")
    public CompletionStage<String> hello(@PathParam("name") String name) {
        return bus.<String>send("greeting", name)                           (1)
                .thenApply(Message::body);                                  (2)
    }
}

src/main/java/org/acme/vertx/GreetingService.java

package org.acme.vertx;
import io.quarkus.vertx.ConsumeEvent;
import javax.enterprise.context.ApplicationScoped;
@ApplicationScoped
public class GreetingService {
    @ConsumeEvent("greeting")
    public String greeting(String name) {
        return "Hello " + name;
    }
}

This bean receives the name, and returns the greeting message.

Now, open your browser to http://localhost:8080/async/Quarkus, and you should see:

Hello Quarkus

To better understand, let’s detail how the HTTP request/response has been handled:

• The request is received by the hello method

• a message containing the name is sent to the event bus

• Another bean receives this message and computes the response

• This response is sent back using the reply mechanism

• Once the reply is received by the sender, the content is written to the HTTP response

This application can be packaged using:

You can also compile it as a native executable with: