[[MicroProfile_Reactive_Messaging_SmallRye]]
= MicroProfile Reactive Messaging Subsystem Configuration
----
:smallrye-reactive-messaging-version:       3.6
:smallrye-reactive-messaging-tag:           {smallrye-reactive-messaging-version}.0
:eclipse-mp-reactive-messaging-api-version: 2.0
----


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 {eclipse-mp-reactive-messaging-api-version}, 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.

=== Programming model and Limitations
See the spec for more thorough examples, this section just attempts to summarize the highlights.

Version 1.0 of the MicroProfile Reactive Messaging specification introduced the `@Incoming` and `@Outgoing` annotations. They are intended for use in an `@ApplicationScoped` (or `@Dependent`) CDI bean:

[source, java]
----
@ApplicationScoped
public class MyBean {
    @Outgoing("in-memory")
    public String generate() {
        return ...; // Do some generation of values
    }

    @Incoming("in-memory")
    public void consume(String value) {
        System.out.println(value);
    }
}
----

Values generated by the `generate()` method will be received by the `consume()` method. In this basic setup where the channel names match, the streams are dealt with in-memory. We'll see how to have them handled by Kafka later on.

In the above example, we are essentially generating values and consuming them with no user-interaction. MicroProfile Reactive Messaging 2.0 introduces a `@Channel` annotation, which can be used to inject a `Publisher` for receiving values sent on streams, and an `Emitter` which can be used to send values to a stream. This makes it easier to send/receive values from code paths resulting from user interaction:

[source, java]
----
@ApplicationScoped
public class MyBean {
    @Inject
    @Channel("in-memory")
    Emitter<String> emitter;

    @Inject
    @Channel("in-memory")
    Publisher<String> publisher;

    void send(String value) {
        emitter.send(value);
    }
}
----


In the above example we can now easily send data to the Reactive Messaging streams by calling `Emitter.send()`. Similarly, we can subscribe to the Publisher and receive the data. However, receiving still has a few shortcomings:

* The above example will not work out of the box. When trying to send on the `Emitter`, you will get an error that there are no subscibers (which in turn runs the risk of causing overflow). This can be worked around by creating a subscription on the `Publisher`.
* At present there can only be one subscription on the injected `Publisher`.

The above points means that this `Publisher` is not usable directly as an asynchronous return value for e.g. a Jakarta RESTFul Webservices endpoint. As the Jakarta RESTFul Webservices request is what will create the subscription such a call would need to happen before calling `Emitter.send()`.

If we replace the `Emitter` with the `generate()` method from the original method our example will work. However, if we return the Publisher to more than one Jakarta RESTFul Webservices request, we end up with more than one subscriptions which will not all receive every single value.

User's applications intended to return published values to users via e.g. Jakarta RESTFul Webservices will need to do their own subscriptions and buffering of the data. Care must be taken to not let the cache grow uncontrolled, which could cause OutOfMemoryErrors.

[[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, java, 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. The same happens for other configuration properties. See the Kafka documentation for more details about how to configure Kafka consumers and producers.


===== Connecting to secure Kafka
If connecting to a Kafka instance secured with SSL and SASL, the following example 'microprofile-config.properties' will help you get started. There are a few new properties. We are showing them on the connector level but they could equally well be defined on the channel level (i.e. with the `mp.messaging.outgoing.to-kafka.` and `mp.messaging.incoming.from-kafka.` prefixes from the previous examples rather than the connector-wide `mp.messaging.connector.smallrye-kafka` prefix).

[source]
----
mp.messaging.connector.smallrye-kafka.bootstrap.servers=localhost:9092
mp.messaging.connector.smallrye-kafka.sasl.mechanism=PLAIN
mp.messaging.connector.smallrye-kafka.security.protocol=SASL_SSL
mp.messaging.connector.smallrye-kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="${USER}" \
  password="${PASSWORD}";
mp.messaging.connector.smallrye-kafka.wildfly.elytron.ssl.context=test

# Channel configuration would follow here, but is left out for brevity
----
Each of these lines has the following meaning:

* `mp.messaging.connector.smallrye-kafka.bootstrap.servers=localhost:9092` - specifies the Kafka servers to connect to. This is the same as in the previous examples
* `mp.messaging.connector.smallrye-kafka.sasl.mechanism=PLAIN` - specifies the SASL mechanism to use. See `sasl.mechanism` in the Kafka documentation for other choices.
* `mp.messaging.connector.smallrye-kafka.security.protocol` - specifies the protocol mechanism to use. See `security.protocol` in the Kafka documentation for other choices. In this case we are using `SASL_SSL` which means that communication is over SSL, and that SASL is used to authenticate
* `mp.messaging.connector.smallrye-kafka.sasl.jaas.config=...` - specifies how we will authenticate with Kafka. In order to not hardcode the credentials in our `microprofile-config.properties file` we are using the property substitution feature of MicroProfile Config. In this case, if you have defined the `USER` and `PASSWORD` environment variables they will be passed in as part of the configuration
* `mp.messaging.connector.smallrye-kafka.wildfly.elytron.ssl.context=test` - this is not needed if Kafka is secured with a CA signed certificate. If you are using self-signed certificates, you will need to specify a truststore in the Elytron subsystem, and create an `SSLContext` referencing that. The value of this property is used to look up the `SSLContext` in the Elytron subsystem under `/subsystem=elytron/client-ssl-context=*` in the WildFly management model. In this case the property value is `test`, so we look up the `SSLContext` defined by `/subsystem=elytron/client-ssl-context=test` and use that configure the truststore to use for the connection to Kafka.

===== Kafka User API
In order to be able to get more information about messages received from Kafka, and to be able to influence how Kafka handles messages, there is a user API for Kafka. This API lives in the https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api[`io/smallrye/reactive/messaging/kafka/api`] package.

The API consists of the following classes:

* https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api/IncomingKafkaRecordMetadata.java[`IncomingKafkaRecordMetadata`] - This metadata contains information such as:
** the `key` of the Kafka record represented by a `Message`
** the Kafka `topic` and `partition` used for the `Message`, and the `offset` within those
** the `Message` `timestamp` and `timestampType`
** the `Message` `headers` - these are pieces of information the application can attach on the producing side, and receive on the consuming side. They are stored and forwarded on by Kafka but have no meaning to Kafka itself.
* https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api/OutgoingKafkaRecordMetadata.java[`OutgoingKafkaRecordMetadata`] - This is constructed via the builder returned via the `builder()` method, and allows you to specify/override how Kafka will handle the messages. Similar to the `IncomingKafkaRecordMetadata` case, you can set:
** the `key`. Kafka will then treat this entry as the key of the message
** the `topic`, as already seen we typically use the `microprofile-config.properties` configuration to specify the topic to use for a channel backed by Kafka. However, in some cases the code sending the message might need to make some choices (for example depending on values contained in the data) about which topic to send to. Specifying this here will make Kafka use that topic.
** the `partition`. Generally, it is best to let Kafka's partitioner choose the partition, but for cases where it is essential to be able to specify it this can be done
** the `timestamp` if you don't want the one auto-generated by Kafka
** `headers` - you can attach headers for the consumer, as mentioned for `IncomingKafkaRecordMetadata`
* https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api/KafkaMetadataUtil.java[`KafkaMetadataUtil`] contains utility methods to write `OutgoingKafkaRecordMetadata` to a `Message`, and to read `IncomingKafkaRecordMetadata` from a `Message`. Note that if you write `OutgoingKafkaRecordMetadata` to a `Message` which is sent to a channel not handled by Kafka it will be ignored, and if you attempt to read `IncomingKafkaRecordMetadata` from a `Message` arriving from a channel no handled by Kafka it will be `null`.

The following example shows how to write and read the `key` from a message:

[source, java, options="nowrap"]
----
@Inject
@Channel("from-user")
Emitter<Integer> emitter;

@Incoming("from-user")
@Outgoing("to-kafka")
public Message<Integer> send(Message<Integer> msg) {
    // Set the key in the metadata
    OutgoingKafkaRecordMetadata<String> md =
            OutgoingKafkaRecordMetadata.<String>builder()
                .withKey("KEY-" + i)
                .build();
    // Note that Message is immutable so the copy returned by this method
    // call is not the same as the parameter to the method
    return KafkaMetadataUtil.writeOutgoingKafkaMetadata(msg, md);
}

@Incoming("from-kafka")
public CompletionStage<Void> receive(Message<Integer> msg) {
    IncomingKafkaRecordMetadata<String, Integer> metadata =
        KafkaMetadataUtil.readIncomingKafkaMetadata(msg).get();

    // We can now read the Kafka record key
    String key = metadata.getKey();

    // When using the Message wrapper around the payload we need to explicitly ack
    // them
    return msg.ack();
}
----
To configure the Kafka mapping we need a `microprofile-config.properties`
```
kafka.bootstrap.servers=kafka:9092

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

mp.messaging.incoming.from-kafka.connector=smallrye-kafka
mp.messaging.incoming.from-kafka.topic=some-topic
mp.messaging.incoming.from-kafka.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
mp.messaging.incoming.from-kafka.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
```
This configuration looks a lot like the previous configuration that we saw, but note that we need to specify the `key.serializer` for the outgoing channel, and the `key.deserializer` for the incoming channel. As before, they are implementations of `org.apache.kafka.common.serialization.Serializer` and `org.apache.kafka.common.serialization.Deserializer` respectively. Kafka provides implementations for basic types, and you may write your own and include them in the deployment.

===== A note on `org.apache.kafka` classes
While we do expose the Kafka Clients jar in our BOMs, its usage is limited to

* Classes/interfaces exposed via the Kafka User API, e.g.:
** `org.apache.kafka.common.header.Header` and `org.apache.kafka.common.header.Headers` and implementations of those that are considered public API as per the Apache Kafka documentation.
** `org.apache.kafka.clients.consumer.ConsumerRecord`
** `org.apache.kafka.common.record.TimestampType`
* Classes/interfaces needed for serialization and deserialization:
** `org.apache.kafka.common.serialization.Deserializer`
** `org.apache.kafka.common.serialization.Serializer`
** Implementatations of `org.apache.kafka.common.serialization.Deserializer` and `org.apache.kafka.common.serialization.Serializer` in the `org.apache.kafka.common.serialization` package


== 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]
****
