////
This guide is maintained in the main Quarkus repository
and pull requests should be submitted there:
https://github.com/quarkusio/quarkus/tree/main/docs/src/main/asciidoc
////
= Using the event bus

include::./attributes.adoc[]

Quarkus allows different beans to interact using asynchronous events, thus promoting loose-coupling.
The messages are sent to _virtual addresses_.
It offers 3 types of delivery mechanism:

- 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.

NOTE: The asynchronous message passing feature allows replying to messages which is not supported by Reactive Messaging.
However, it is limited to single-event behavior (no stream) and to local messages.

== Installing

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

:create-app-artifact-id: vertx-quickstart
:create-app-extensions: vertx,resteasy-mutiny
include::includes/devtools/create-app.adoc[]

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

:add-extension-extensions: vertx
include::includes/devtools/extension-add.adoc[]

Otherwise, you can manually add this to the dependencies section of your build file:

[source,xml,role="primary asciidoc-tabs-target-sync-cli asciidoc-tabs-target-sync-maven"]
.pom.xml
----
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-vertx</artifactId>
</dependency>
----

[source,gradle,role="secondary asciidoc-tabs-target-sync-gradle"]
.build.gradle
----
implementation("io.quarkus:quarkus-vertx")
----

== Consuming events

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

[source, java]
----
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();
    }
}
----
<1> If not set, the address is the fully qualified name of the bean, for instance, in this snippet it's `org.acme.vertx.GreetingService`.
<2> The method parameter is the message body. If the method returns _something_ it's the message response.

[IMPORTANT]
====
By default, the code consuming the event must be _non-blocking_, as it's called on the Vert.x event loop.
If your processing is blocking, use the `blocking` attribute:

[source, java]
----
@ConsumeEvent(value = "blocking-consumer", blocking = true)
void consumeBlocking(String message) {
    // Something blocking
}
----

Alternatively, you can annotate your method with `@io.smallrye.common.annotation.Blocking`:
[source, java]
----
@ConsumeEvent(value = "blocking-consumer")
@Blocking
void consumeBlocking(String message) {
    // Something blocking
}
----

When using `@Blocking`, it ignores the value of the `blocking` attribute of `@ConsumeEvent`.
See the xref:quarkus-reactive-architecture.adoc[Quarkus Reactive Architecture documentation] for further details on this topic.
====

Asynchronous processing is also possible by returning either an `io.smallrye.mutiny.Uni` or a `java.util.concurrent.CompletionStage`:

[source,java]
----
package org.acme.vertx;

import io.quarkus.vertx.ConsumeEvent;

import javax.enterprise.context.ApplicationScoped;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import io.smallrye.mutiny.Uni;

@ApplicationScoped
public class GreetingService {

    @ConsumeEvent
    public CompletionStage<String> consume(String name) {
        // return a CompletionStage completed when the processing is finished.
        // You can also fail the CompletionStage explicitly
    }

    @ConsumeEvent
    public Uni<String> process(String name) {
        // return an Uni completed when the processing is finished.
        // You can also fail the Uni explicitly
    }
}
----

[TIP]
.Mutiny
====
The previous example uses Mutiny reactive types.
If you are not familiar with Mutiny, check xref:mutiny-primer.adoc[Mutiny - an intuitive reactive programming library].
====

=== Configuring the address

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

[source, java]
----
@ConsumeEvent("greeting")               // <1>
public String consume(String name) {
    return name.toUpperCase();
}
----
<1> Receive the messages sent to the `greeting` address

=== Replying

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.

[source, java]
----
@ConsumeEvent("greeting")
public String consume(String name) {
    return name.toUpperCase();
}
----

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

[source, java]
----
@ConsumeEvent("greeting")
public Uni<String> consume2(String name) {
    return Uni.createFrom().item(() -> name.toUpperCase()).emitOn(executor);
}
----

[NOTE]
====
You can inject an `executor` if you use the Context Propagation extension:
[source, java]
----
@Inject ManagedExecutor executor;
----

Alternatively, you can use the default Quarkus worker pool using:

[source, java]
----
Executor executor = Infrastructure.getDefaultWorkerPool();
----
====

=== Implementing fire and forget interactions

You don't have to reply to received messages.
Typically, for a _fire and forget_ interaction, the messages are consumed and the sender does not need to know about it.
To implement this, your consumer method just returns `void`

[source,java]
----
@ConsumeEvent("greeting")
public void consume(String event) {
    // Do something with the event
}
----

=== Dealing with messages

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

[source, java]
----
@ConsumeEvent("greeting")
public void consume(Message<String> msg) {
    System.out.println(msg.address());
    System.out.println(msg.body());
}
----

=== Handling Failures

If a method annotated with `@ConsumeEvent` throws an exception then:

* if a reply handler is set then the failure is propagated back to the sender via an `io.vertx.core.eventbus.ReplyException` with code `ConsumeEvent#FAILURE_CODE` and the exception message,
* if no reply handler is set then the exception is rethrown (and wrapped in a `RuntimeException` if necessary) and can be handled by the default exception handler, i.e. `io.vertx.core.Vertx#exceptionHandler()`.

== Sending messages

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:

[source, java]
----
package org.acme.vertx;

import io.smallrye.mutiny.Uni;
import io.vertx.mutiny.core.eventbus.EventBus;
import io.vertx.mutiny.core.eventbus.Message;
import org.jboss.resteasy.annotations.jaxrs.PathParam;

import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

@Path("/async")
public class EventResource {

    @Inject
    EventBus bus;                                       // <1>

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("{name}")
    public Uni<String> greeting(@PathParam String name) {
        return bus.<String>request("greeting", name)        // <2>
                .onItem().transform(Message::body);
    }
}
----
<1> Inject the Event bus
<2> Send a message to the address `greeting`. Message payload is `name`

The `EventBus` object provides methods to:

1. `send` a message to a specific address - one single consumer receives the message.
2. `publish` a message to a specific address - all consumers receive the messages.
3. `send` a message and expect reply

[source, java]
----
// Case 1
bus.sendAndForget("greeting", name)
// Case 2
bus.publish("greeting", name)
// Case 3
Uni<String> response = bus.<String>request("address", "hello, how are you?")
        .onItem().transform(Message::body);
----

== Putting things together - bridging HTTP and messages

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:

:create-app-artifact-id: vertx-http-quickstart
:create-app-extensions: vertx,resteasy-mutiny
include::includes/devtools/create-app.adoc[]

You can already start the application in _dev mode_ using:

include::includes/devtools/dev.adoc[]

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

[source,java]
.src/main/java/org/acme/vertx/EventResource.java
----
package org.acme.vertx;

import io.smallrye.mutiny.Uni;
import io.vertx.mutiny.core.eventbus.EventBus;
import io.vertx.mutiny.core.eventbus.Message;
import org.jboss.resteasy.annotations.jaxrs.PathParam;

import javax.inject.Inject;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;

@Path("/async")
public class EventResource {

    @Inject
    EventBus bus;

    @GET
    @Produces(MediaType.TEXT_PLAIN)
    @Path("{name}")
    public Uni<String> greeting(@PathParam String name) {
        return bus.<String>request("greeting", name)            // <1>
                .onItem().transform(Message::body);            // <2>
    }
}
----
<1> send the `name` to the `greeting` address and request a response
<2> when we get the response, extract the body and send it to the user

If you call this endpoint, you will wait and get a timeout. Indeed, no one is listening.
So, we need a consumer listening on the `greeting` address. Create a `GreetingService` bean with the following content:

[source, java]
.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:

[source,text]
----
Hello Quarkus
----

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

1. The request is received by the `hello` method
2. a message containing the _name_ is sent to the event bus
3. Another bean receives this message and computes the response
4. This response is sent back using the reply mechanism
5. Once the reply is received by the sender, the content is written to the HTTP response

This application can be packaged using:

include::includes/devtools/build.adoc[]

You can also compile it as a native executable with:

include::includes/devtools/build-native.adoc[]

== Using codecs

The https://vertx.io/docs/vertx-core/java/#event_bus[Vert.x Event Bus] uses codecs to _serialize_ and _deserialize_ objects.
Quarkus provides a default codec for local delivery.
So you can exchange objects as follows:

[source, java]
----
@GET
@Produces(MediaType.TEXT_PLAIN)
@Path("{name}")
public Uni<String> greeting(@PathParam String name) {
    return bus.<String>request("greeting", new MyName(name))
        .onItem().transform(Message::body);
}

@ConsumeEvent(value = "greeting")
Uni<String> greeting(MyName name) {
    return Uni.createFrom().item(() -> "Hello " + name.getName());
}
----

If you want to use a specific codec, you need to explicitly set it on both ends:

[source, java]
----
@GET
@Produces(MediaType.TEXT_PLAIN)
@Path("{name}")
public Uni<String> greeting(@PathParam String name) {
    return bus.<String>request("greeting", name,
        new DeliveryOptions().setCodecName(MyNameCodec.class.getName())) // <1>
        .onItem().transform(Message::body);
}

@ConsumeEvent(value = "greeting", codec = MyNameCodec.class)            // <2>
Uni<String> greeting(MyName name) {
    return Uni.createFrom().item(() -> "Hello "+name.getName());
}
----
<1> Set the name of the codec to use to send the message
<2> Set the codec to use to receive the message
