WIXLIB

MessageAt the core of the Reactive Messaging specification is the concept of message. A message is anenvelope wrapping a payload. A message is sent to a specific channel and, when received andprocessed successfully, acknowledged.Reactive Messaging application components are addressable recipients which await the arrival ofmessages on a channel and react to them, otherwise lying dormant.Messages are represented by the e class. Thisinterface is intentionally kept minimal. The aim is that connectors will provide their ownimplementations with additional metadata that is relevant to that connector. For instance, aKafkaMessage would provide access to the topic and g.Message#ackthemethodacknowledges the message. Note that the ack method is asynchronous as acknowledgement isgenerally an asynchronous process.Plain messages are created using: e#of(T) - wraps the given payload, noacknowledgement e#of(T,java.util.function.Supplier java.util.concurrent.CompletionStage java.lang.Void ) - wrapsthe given payload and provides the acknowledgment logicMessage consumption with @IncomingThe ng annotation is used on a method from aCDI bean to indicate that the method consumes messages from the specified channel:@Incoming("my-channel")public CompletionStage Void consume(Message String message) {return message.ack();}①②1. my-channel is the channel2. the method is called for every message sent to the my-channel channelReactive Messaging supports various forms of method signatures. This is detailed in the nextsection.Remember that Reactive Messaging interactions are assembled from Reactive Streams. A methodannotated with @Incoming is a Reactive Streams subscriber and so consumes messages that fit withthe message signature and its annotations. Note that the handling of the Reactive Streams protocol,such as subscriptions and back pressure, is managed by the Reactive Messaging implementation.The MicroProfile Reactive Streams specification used as a foundation for this version of Reactive7

Messaging is a single subscriber model where a stream Publisher is connected to a single Subscriberwhich controls back pressure. This implies that a Reactive Messaging channel should appear in asingle @Incoming annotation. The annotation of more than one @Incoming method to be associatedwith the same channel is not supported and will cause an error during deployment.From the user perspective, whether the incoming messages comes from co-located beans or aremote message broker is transparent. However, the user may decide to consume a specificsubclass of Message (e.g. KafkaMessage in the following example) if the user is aware of ic CompletionStage Void consume(KafkaMessage String message) {return message.ack();}①1. Explicit consumption of a KafkaMessageMessage production with @OutgoingThe ng annotation is used to annotate a methodfrom a CDI bean to indicate that the method publishes messages to a specified channel:@Outgoing("my-channel")public Message String publish() {return Message.of("hello");}①②③1. my-channel is the targeted channel2. the method is called for every consumer request3. you can create a plain ge#of(T)usingReactive Messaging supports various forms of method signatures. This is detailed in the nextsection.A method annotated with @Outgoing is a Reactive Streams publisher and so publishes messagesaccording to the requests it receives. The downstream @Incoming method or outgoing connectorwith a matching channel name will be linked to this publisher. Only a single method can beannotated with @Outgoing for a particular channel name. Having the same channel name in morethan one @Outgoing annotated method is not supported and will result in an error duringdeployment.Method consuming and producingA method can combine the @Incoming and @Outgoing annotation and will then act as a ReactiveStreams processor:8

oing-channel")public Message String process(Message String message) {return �②1. The incoming channel2. The outgoing channelHaving the same channel appear in the @Outgoing and @Incoming annotations of a processor is notsupported and will result in an error during deployment.ConnectorsThe application can receive and forward messages from various message brokers or transportlayers. For instance, an application can be connected to a Kafka cluster, an AMQP broker or anMQTT server.Reactive Messaging Connectors are extensions managing the communication with a specifictransport technology. They are responsible for mapping a specific channel to remote sink or sourceof messages. This mapping is configured in the application configuration. Note that animplementation may provide various ways to configure the mapping, but support for MicroProfileConfig as a configuration source is mandatory.Connector implementations are associated with a name corresponding to a messaging transport,such as Apache Kafka, Amazon Kinesis, RabbitMQ or Apache ActiveMQ. For instance, anhypothetical Kafka connector could be associated with the following name: acme.kafka. This name isindicated using a qualifier on the connector implementation.The user can associate a channel with this connector using the associated r acme.kafka①1. the name associated with the connector.The configuration format is detailed later in this document.The Reactive Messaging implementation is responsible for finding the connector implementationassociated with the given name in the user configuration. If the connector cannot be found, thedeployment of the application must be failed.The Reactive Messaging specification provides an SPI to implement connectors.Message stream operationMessage stream operation occurs according to the principles of reactive programming. The backpressure mechanism of reactive streams means that a publisher will not send data to a subscriber9

unless there are outstanding subscriber requests. This implies that data flow along the stream isenabled by the first request for data received by the publisher. For methods that are annotated with@Incoming and @Outgoing this data flow control is handled automatically by the underlying systemwhich will call the @Incoming and @Outgoing methods as appropriate.Although @Incoming and @Outgoing methods remain callable from Java code, callingthem directly will not affect the reactive streams they are associated with. Forexample, calling an @Outgoing annotated method from user code will not post amessage on a message queue and calling an @Incoming method cannot be used toNOTEread a message. Enabling this would bypass the automatic back pressuremechanism that is one of the benefits of the specification. The @Incoming and@Outgoing method annotations are used to declaratively define the stream which isthen run by the implementation of MicroProfile Reactive Messaging without theuser’s code needing to handle concerns such as subscriptions or flow control withinthe stream.Supported CDI scopesImplementations of the Reactive Messaging specification must support at least the following CDIscopes: @ApplicationScoped beans @Dependent beansThe following code gives an example of a bean annotated with @ApplicationScoped:@ApplicationScopedpublic class ApplicationScopeBeans {@Outgoing("source")public Publisher Integer source() {return e")@Outgoing("output")public int process(int i) {return i 1;}@Incoming("output")public void sink(int v) {System.out.println(v);}}Implementations can provide support for other scopes. However the behavior is not defined.10

Supported method signaturesThe signature of message stream methods can have a number of different distinct types, offeringdiffering levels of power and simplicity to application developers. Different shapes are supporteddepending on whether the method is a publisher, subscriber or processor, for example, apublishing stream supports returning MicroProfile Reactive Streams PublisherBuilder, but notSubscriberBuilder, the inverse is true for a subscribing stream.This section lists the methods signatures that must be supported by the Reactive Messagingimplementation. Implementations must validate that the stream shape matches the @Outgoing and@Incoming annotations, if they don’t, a CDI definition exception should be raised to the CDI containerduring initialization.It’s important to remember that users must not call these methods directly. They are invoked by theReactive Messaging implementation following the Reactive Streams protocol.Also the method must be implemented in a non-blocking fashion. For blocking transformations,asynchronous variants can be used.NOTENOTEassembly time is when the Reactive Messaging implementation initializes itself andcreates the different bean instances and connects them together.In the following lists, Message can be an implementation of the Message interface.Methods producing dataSignature@Outgoing("name")Publisher Message O method()@Outgoing("channel")Publisher O method()@Outgoing("channel")PublisherBuilder Message O method()BehaviorInvocationReturns a stream ofMethod called once atMessage associated with assembly time.the channel name.Returns a stream ofpayload of type Oassociated with thechannel channel.Produced payloads aremapped to Message O by the ReactiveMessagingimplementation.Method called once atassembly time.Returns a stream ofMethod called once atMessage associated with assembly time.the channel channel.11

Signature@Outgoing("channel")PublisherBuilder O method()@Outgoing("channel")Message O method()@Outgoing("channel")O method()@Outgoing("channel")CompletionStage Message O method()12BehaviorInvocationReturns a stream ofMethod called once atpayload associated with subscription time.the channel channel.Produced payloads aremapped to Message O by the ReactiveMessagingimplementation.Produces an infinitestream of Messageassociated with thechannel channel.This method is calledfor each request madeby the subscriber.Produces an infinitestream of payloadassociated with thechannel channel.Produced payloads aremapped to Message O by the ReactiveMessagingimplementation.This method is calledfor each request madeby the subscriber.Produces an infinitestream of Messageassociated with thechannel channel. Theresult is aCompletionStage. Themethod should not becalled by the reactivemessagingimplementation untilthe CompletionStagereturned previously iscompleted.This method is calledfor each request madeby the subscriber.

Signature@Outgoing("channel")CompletionStage O method()BehaviorInvocationProduces an infinitestream of payloadassociated with thechannel channel.Produced payloads aremapped to Message O by the ReactiveMessagingimplementation. Theresult is aCompletionStage. Themethod should not becalled by the reactivemessagingimplementation untilthe CompletionStagereturned previously iscompleted.This method is calledfor each request madeby the subscriber.BehaviorInvocationReturns a Subscriberthat receives theMessage objectstransiting on thechannel channel.The method is calledonly once to retrievethe Subscriber object atassembly time. Thissubscriber is connectedto the matchingchannel.Returns a Subscriberthat receives thepayload objectstransiting on thechannel channel. Thepayload isautomatically extractedfrom the inflightmessages usingMessage.getPayload().The method is calledonly once to retrievethe Subscriber object atassembly time. Thissubscriber is connectedto the matchingchannel.Returns aSubscriberBuilder thatreceives the Messageobjects transiting onthe channel channel.The method is calledonly once at assemblytime to retrieve aSubscriberBuilder thatis used to build aCompletionSubscriberthat is subscribed to thematching channel.Methods consuming dataSignature@Incoming("channel")Subscriber Message I method()@Incoming("channel")Subscriber I method()@Incoming("channel")SubscriberBuilder Message I method()13

Signature@Incoming("channel")SubscriberBuilder I method()@Incoming("channel")void method(I payload)@Incoming("channel")CompletionStage ? method(Message I msg)14BehaviorInvocationReturns aSubscriberBuilder thatis used to build aCompletionSubscriber I that receives thepayload of each Message.The payload isautomatically extractedfrom the inflightmessages usingMessage.getPayload().The method is calledonly once at assemblytime to retrieve aSubscriberBuilder thatis used to build aCompletionSubscriberthat is subscribed to thematching channel.Consumes the payload.The method can returnvoid or any object ornull. The returnedvalue is ignored.This method is calledfor every Message I instance transiting onthe channel channel.The payload isautomatically extractedfrom the inflightmessages usingMessage.getPayload().The user method isnever calledconcurrently and somust return beforebeing called with thenext payload.Consumes the MessageThis method is calledfor every Message I instance transiting onthe channel channel.The user method isnever calledconcurrently. Thereactive messagingimplementation mustwait until thecompletion of thepreviously returnedCompletionStage beforecalling the methodagain with the nextMessage. Note that@Incoming("channel")void method(Message I msg) is not allowed asmessageacknowledgement isasynchronous.

Signature@Incoming("channel")CompletionStage ? method(I payload)BehaviorInvocationConsumes the payloadasynchronouslyThis method is calledfor every Message I instance transiting onthe channel channel.The payload isautomatically extractedfrom the inflightmessages usingMessage.getPayload().The user method isnever calledconcurrently. Thereactive messagingimplementation mustwait until thecompletion of thepreviously returnedCompletionStage beforecalling the methodagain with the nextpayload.BehaviorInvocationReturns a ReactiveStreams processorconsuming incomingMessage instances andproduces Messageinstances.This method is calledonce; at assembly time.Returns a ReactiveStreams processorconsuming incomingpayload instances andproduces payloadinstances.This method is calledonce; at assembly time.Returns aProcessorBuilderconsuming incomingMessage instances andproduces Messageinstances.This method is calledonce; at assembly time.Methods processing sor Message I , Message O method()@Incoming("in")@Outgoing("out")Processor I, O uilder Message I , Message O method();15

uilder I, O method();@Incoming("in")@Outgoing("out")Publisher Message O method(Message I msg)16BehaviorInvocationReturns a ReactiveThis method is calledStreams processor that once; at assembly time.consuming incomingpayload instances andproduces payloadinstances.Returns a ReactiveStreams Publisher foreach incoming Message.The returned Publishercan be empty or emitsmultiple Messageinstances. If thereturned Publisheremits several elements,these elements areflattened in theoutgoing stream as aconcatenation ofelements. The flatteningfollows the samesemantics as theflatMap operator fromthe MicroProfileReactive Streamsspecification.This method is calledfor every incomingmessage.Implementations mustnot call the methodsubsequently until thestream from thepreviously returnedPublisher is completed.

Signature@Incoming("in")@Outgoing("out")Publisher O method(I ilder Message O method(Message I msg)BehaviorInvocationReturns a ReactiveStreams Publisher foreach incoming payload.The returned Publishercan be empty or emitsmultiple payloadinstances. If thereturned Publisheremits several elements,these elements areflattened in theoutgoing stream as aconcatenation ofelements. The flatteningfollows the samesemantics as theflatMap operator fromthe MicroProfileReactive Streamsspecification. TheReactive Messagingimplementation mustcreate new Messageinstances for eachemitted payload as wellas extracing thepayload for eachincoming Message.This method is calledfor every incomingmessage.Implementations mustnot call the methodsubsequently until thestream from thepreviously returnedPublisher is completed.Returns aPublisherBuilder foreach incoming Message.The stream resultingfrom the built Publishercan be empty or emitsmultiple Messageinstances. If the streamemitted from the builtPublisher emits severalelements, theseelements are flattenedin the outgoing streamas a concatenation ofelements. The flatteningfollows the samesemantics as theflatMap operator fromthe MicroProfileReactive Streamsspecification.This method is calledfor every incomingmessage.Implementations mustnot call the methodsubsequently until thestream built from thepreviously returnedPublisherBuilder iscompleted.17

uilder O method(I payload)@Incoming("in")@Outgoing("out")Message O method(Message I msg)@Incoming("in")@Outgoing("out")O method(I payload)18BehaviorInvocationReturns aPublisherBuilder foreach incoming payload.The stream resultingfrom the built Publishercan be can be empty oremits multiple payloadinstances. If the streamemitted from the builtPublisher emits severalelements, theseelements are flattenedin the outgoing streamas a concatenation ofelements. The flatteningfollows the samesemantics as theflatMap operator fromthe MicroProfileReactive Streamsspecification. Th

microservices written using MicroProfile are able to meet the demands required by the architecture. Use cases MicroProfile Reactive Messaging aims to provide a way to connect event-driven microservices. The key characteristics of the specification make it versatile and suitable for building different types of architecture and applications.