[[MicroProfile_Reactive_Messaging_SmallRye]]
= MicroProfile Reactive Messaging Subsystem Configuration
----
:smallrye-reactive-messaging-version:    2.5
----


Support for https://microprofile.io/project/eclipse/microprofile-reactive-messaging[MicroProfile Reactive Messaging] is
provided as a Tech Preview feature by the _microprofile-reactive-messaging-smallrye_ subsystem.

[[required-extension-microprofile-reactive-messaging-smallrye]]
== Required Extension

This extension is *not* included in the standard configurations included in the WildFly distribution.

You can add the extension to a configuration either by adding
an `<extension module="org.wildfly.extension.microprofile.reactive-messaging-smallrye"/>`
element to the xml or by using the following CLI operation:

[source,options="nowrap"]
----
[standalone@localhost:9990 /] /extension=org.wildfly.extension.microprofile.reactive-messaging-smallrye:add
{"outcome" => "success"}

[standalone@localhost:9990 /] /subsystem=microprofile-reactive-messaging-smallrye:add
{
    "outcome" => "success",
    "response-headers" => {
        "operation-requires-reload" => true,
        "process-state" => "reload-required"
    }
}
----

To use this subsystem, you must also enable the <<MicroProfile_Reactive_Streams_Operators_SmallRye, MicroProfile Reactive Streams Operators>> extension and subsystem.

If you provision your own server and include the `microprofile-reactive-messaging` Galleon layer, you will get the required modules, and the extension and subsystem will be added to your configuration.

If you provision the `microprofile-reactive-messaging-kafka` Galleon layer it includes the modules to enable the Kafka connector functionality. The `microprofile-reactive-messaging-kafka` layer includes the `microprofile-reactive-messaging` layer which provides the core MicroProfile Reactive Messaging functionality.

== Specification

WildFly's MicroProfile Reactive Messaging subsystem implements MicroProfile Reactive Messaging 1.0, which adds support for asynchronous messaging support based on MicroProfile Reactive Streams Operators.

== Configuration
The `microprofile-reactive-messaging-smallrye` subsystem contains no configurable attributes or resources. For the core MicroProfile Reactive Messaging functionality there is no configuration. For configuration of the <<microprofile-reactive-messaging-smallrye-config-connectors, connectors>> to external brokers MicroProfile Config is used.

=== Activation
The subsystem will scan all deployments to find classes containing methods with the `org.eclipse.microprofile.reactive.messaging.Incoming` or `org.eclipse.microprofile.reactive.messaging.Outgoing` annotations. If these annotations are found, Reactive Messaging will be enabled for the deployment.

[[microprofile-reactive-messaging-smallrye-config-connectors]]
=== Connectors
MicroProfile Reactive Messaging is designed to be flexible enough to integrate with a wide variety of external messaging systems. This functionality is provided via 'connectors'.

The only included connector at the moment is the Kafka connector.

Connectors are configured using MicroProfile Config. The property keys for the methods have some prefixes mandated by the MicroProfile Reactive Messaging Specification which lists these as:

* `mp.messaging.incoming.[channel-name].[attribute]=[value]`
* `mp.messaging.outgoing.[channel-name].[attribute]=[value]`
* `mp.messaging.connector.[connector-name].[attribute]=[value]`

Essentially `channel-name` is the `@Incoming.value()` or the  `@Outgoing.value()`.

If we have the following pair of methods:

[source,options="nowrap"]
----
@Outgoing("to")
public int send() {
    int i = // Randomly generated...
}

@Incoming("from")
public void receive(int i) {
    // Process payload
}
----
Then the property prefixes mandated by the MicroProfile Reactive Messaging specifications are:

* `mp.messaging.incoming.from.` - this would pick out the property as configuration of the `receive()` method.
* `mp.messaging.outgoing.to.` - this would pick out the property as configuration of the `send()` method.

Note that although these prefixes are understood by the subsystem, the full set depends on the connector you want to configure. Different connectors understand different properties.


==== Kafka Connector

An example of a minimal microprofile-config.properties file for Kafka for the example application shown previously:

```
kafka.bootstrap.servers=kafka:9092

mp.messaging.outgoing.to.connector=smallrye-kafka
mp.messaging.outgoing.to.topic=my-topic
mp.messaging.outgoing.to.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer

mp.messaging.incoming.from.connector=smallrye-kafka
mp.messaging.incoming.from.topic=my-topic
mp.messaging.incoming.from.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
```

Next we will briefly discuss each of these entries. Remember the `to` channel is on the `send()` method, and the `from` channel is on the `receive()` method.

`kafka.bootstrap.servers=kafka:9092` sets the URL of the Kafka broker to connect to for the whole application. It could also be done for just the `to` channel by setting `mp.messaging.outgoing.to.bootstrap.servers=kafka:9092` instead.

`mp.messaging.outgoing.to.connector=smallrye-kafka` says that we want to use Kafka to back the `to` channel. Note that the value `smallrye-kafka` is SmallRye Reactive Messaging specific, and will only be understood if the Kafka connector is enabled.

`mp.messaging.outgoing.to.topic=my-topic` says that we will send data to the Kafka topic called `my-topic`.

`mp.messaging.outgoing.to.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer` tells the connector to use `IntegerSerializer` to serialize the values output by the `send()` method when writing to the topic. Kafka provides serializers for the standard Java types. You may implement your own serializer by writing a class implementing `org.apache.kafka.common.serialization.Serializer` and including it in the deployment.

`mp.messaging.incoming.from.connector=smallrye-kafka` says that we want to use Kafka to back the `from` channel. As above, the value `smallrye-kafka` is SmallRye Reactive Messaging specific.

`mp.messaging.incoming.from.topic=my-topic` says that we will read data from the Kafka topic called `my-topic`.


`mp.messaging.incoming.from.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer` tells the connector to use `IntegerDeserializer` to deserialize the values from the topic before calling the `receive()` method. You may implement your own deserializer by writing a class implementing `org.apache.kafka.common.serialization.Deserializer` and including it in the deployment.

In addition to the above, Apache Kafka, and SmallRye Reactive Messaging's Kafka connector understand a lot more properties. These can be found in the SmallRye Reactive Messaging Kafka connector https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/{smallrye-reactive-messaging-version}/kafka/kafka.html[documentation], and in the Apache Kafka documentation for the https://kafka.apache.org/documentation/#producerconfigs[producers] and the https://kafka.apache.org/documentation/#consumerconfigs[consumers].

The prefixes discussed above are stripped off before passing the property to Kafka.

== Component Reference

The MicroProfile Reactive Messaging implementation is provided by the SmallRye Reactive Messaging project.

****
* https://github.com/eclipse/microprofile-reactive-messaging[MicroProfile Reactive Messaging]
* https://github.com/smallrye/smallrye-reactive-messaging[SmallRye Reactive Messaging]
****
