From f93a740065c0847c4666b0b9b54d98217e4cc7f0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 14 Jan 2020 16:17:40 -0500 Subject: [PATCH] Messaging Annotations: process ReactiveMH (#3141) * Messaging Annotations: process ReactiveMH * Add support for `ReactiveMessageHandler` as a `@ServiceActivator` bean * Wrap `ReactiveMessageHandler` in the `ReactiveMessageHandlerAdapter` in the `ServiceActivatorAnnotationPostProcessor` * Unwrap `ReactiveMessageHandlerAdapter` in the `AbstractMethodAnnotationPostProcessor` for `ReactiveStreamsConsumer` * * Add documentation for the `ReactiveMessageHandler` & Reactive MongoDb channel adapters --- ...AbstractMethodAnnotationPostProcessor.java | 29 +++++++---- ...rviceActivatorAnnotationPostProcessor.java | 9 +++- ...ingAnnotationsWithBeanAnnotationTests.java | 51 +++++++++++++++++-- src/reference/asciidoc/mongodb.adoc | 26 ++++++++++ src/reference/asciidoc/reactive-streams.adoc | 14 ++++- src/reference/asciidoc/whats-new.adoc | 14 ++++- 6 files changed, 124 insertions(+), 19 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index db5157e10c..9e809a7141 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -66,6 +66,7 @@ import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.LambdaMessageProcessor; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.router.AbstractMessageRouter; @@ -104,7 +105,7 @@ public abstract class AbstractMethodAnnotationPostProcessor messageHandlerAttributes = new ArrayList<>(); // NOSONAR @@ -156,16 +157,18 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { AbstractReplyProducingMessageHandler serviceActivator; if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) { - final Object target = resolveTargetBeanFromMethodWithBeanAnnotation(method); + Object target = resolveTargetBeanFromMethodWithBeanAnnotation(method); serviceActivator = extractTypeIfPossible(target, AbstractReplyProducingMessageHandler.class); if (serviceActivator == null) { + if (target instanceof ReactiveMessageHandler) { + return new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) target); + } if (target instanceof MessageHandler) { /* * Return a reply-producing message handler so that we still get 'produced no reply' messages diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java index 0ee68dfc04..8a6a4e26dd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,8 +29,7 @@ import java.util.stream.Stream; import javax.annotation.Resource; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -56,6 +55,7 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.EnableMessageHistory; @@ -73,6 +73,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.support.ErrorMessage; @@ -80,7 +81,11 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import reactor.core.publisher.Mono; +import reactor.core.publisher.MonoProcessor; +import reactor.test.StepVerifier; /** * @author Artem Bilan @@ -90,7 +95,7 @@ import org.springframework.test.context.junit4.SpringRunner; * @since 4.0 */ @ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class) -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class MessagingAnnotationsWithBeanAnnotationTests { @@ -225,6 +230,25 @@ public class MessagingAnnotationsWithBeanAnnotationTests { .withMessageContaining("The attribute causing the ambiguity is: [applySequence]."); } + @Autowired + private MessageChannel reactiveMessageHandlerChannel; + + @Autowired + private ContextConfiguration contextConfiguration; + + @Test + public void testReactiveMessageHandler() { + this.reactiveMessageHandlerChannel.send(new GenericMessage<>("test")); + + StepVerifier.create( + this.contextConfiguration.messageMonoProcessor + .map(Message::getPayload) + .cast(String.class)) + .expectNext("test") + .verifyComplete(); + } + + @Configuration @EnableIntegration @EnableMessageHistory @@ -403,6 +427,23 @@ public class MessagingAnnotationsWithBeanAnnotationTests { return collector()::add; } + MonoProcessor> messageMonoProcessor = MonoProcessor.create(); + + @Bean + MessageChannel reactiveMessageHandlerChannel() { + return new FluxMessageChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "reactiveMessageHandlerChannel") + public ReactiveMessageHandler reactiveMessageHandlerService() { + return (message) -> { + messageMonoProcessor.onNext(message); + messageMonoProcessor.onComplete(); + return Mono.empty(); + }; + } + } @Configuration diff --git a/src/reference/asciidoc/mongodb.adoc b/src/reference/asciidoc/mongodb.adoc index 556058d614..9c909b4d4c 100644 --- a/src/reference/asciidoc/mongodb.adoc +++ b/src/reference/asciidoc/mongodb.adoc @@ -467,3 +467,29 @@ private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() { } ---- ==== + +[[mongodb-reactive-channel-adapters]] +=== MongoDB Reactive Channel Adapters + +Starting with version 5.3, the `ReactiveMongoDbStoringMessageHandler` implementation is provided. +It is based on the `ReactiveMongoOperations` from Spring Data and requires a `org.mongodb:mongodb-driver-reactivestreams` dependency. +This is an implementation of the `ReactiveMessageHandler` which is supported natively in the framework when reactive streams composition is involved in the integration flow definition. +See more information in the <<./reactive-streams.adoc/reactive-message-handler,ReactiveMessageHandler>>. + +From configuration perspective there is no difference with many other standard channel adapters. +For example with Java DSL such a channel adapter could be used like: + +==== +[source, java] +---- +@Bean +public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbFactory) { + return f -> f + .channel(MessageChannels.flux()) + .handle(MongoDb.reactiveOutboundChannelAdapter(mongoDbFactory)); +} +---- +==== + +In this sample we are going to connect to the MongoDb via provided `ReactiveMongoDatabaseFactory` and store a data from request message into a default collection with the `data` name. +The real operation is going to be performed on-demand from the reactive stream composition in the internally created `ReactiveStreamsConsumer`. diff --git a/src/reference/asciidoc/reactive-streams.adoc b/src/reference/asciidoc/reactive-streams.adoc index 3b9bbf8710..c65cbc621b 100644 --- a/src/reference/asciidoc/reactive-streams.adoc +++ b/src/reference/asciidoc/reactive-streams.adoc @@ -105,6 +105,18 @@ A result of the function is wrapped into a `Mono>` for flat-mapping i See <<./dsl.adoc#java-dsl,Java DSL Chapter>> for more information. +[[reactive-message-handler]] +=== `ReactiveMessageHandler` + +Starting with version 5.3, the `ReactiveMessageHandler` is supported natively in the framework. +This type of message handler is designed for reactive clients which return a reactive type for on-demand subscription for low-level operation execution and doesn't provide any reply data to continue a reactive stream composition. +When a `ReactiveMessageHandler` is used in the imperative integration flow, the `handleMessage()` result in subscribed immediately after return, just because there is no reactive streams composition in such a flow to honor back-pressure. +In this case the framework wraps this `ReactiveMessageHandler` into a `ReactiveMessageHandlerAdapter` - a plain implementation of `MessageHandler`. +However when a `ReactiveStreamsConsumer` is involved in the flow (e.g. when channel to consume is a `FluxMessageChannel`), such a `ReactiveMessageHandler` is composed to the whole reactive stream with a `flatMap()` Reactor operator to honor back-pressure during consumption. + +One of the out-of-the-box `ReactiveMessageHandler` implementation is a `ReactiveMongoDbStoringMessageHandler` for Outbound Channel Adapter. +See <<./mongodb.adoc#mongodb-reactive-channel-adapters,MongoDB Reactive Channel Adapters>> for more information. + [[reactive-channel-adapters]] === Reactive Channel Adapters @@ -120,7 +132,7 @@ A reactive outbound channel adapter implementation is about initiation (or conti An inbound payload could be a reactive type per se or as an event of the whole integration flow which is a part of reactive stream on top. A returned reactive type can be subscribed immediately if we are in one-way, fire-and-forget scenario, or it is propagated downstream (request-reply scenarios) for further integration flow or an explicit subscription in the target business logic, but still downstream preserving reactive streams semantics. -Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>> and <<./rsocket.adoc#rsocket,RSocket>>. +Currently Spring Integration provides channel adapter (or gateway) implementations for <<./webflux.adoc#webflux,WebFlux>>, <<./rsocket.adoc#rsocket,RSocket>> and <<./mongodb.adoc#mongodb,MongoDb>>. Also an https://github.com/spring-projects/spring-integration-extensions/tree/master/spring-integration-cassandra[Apache Cassandra Extension] provides a `MessageHandler` implementation for the Cassandra reactive driver. More reactive channel adapters are coming, for example for https://r2dbc.io/[R2DBC], https://mongodb.github.io/mongo-java-driver-reactivestreams/[MongoDB], for Apache Kafka in https://github.com/spring-projects/spring-integration-kafka[Spring Integration Kafka] based on the `ReactiveKafkaProducerTemplate` and `ReactiveKafkaConsumerTemplate` from https://spring.io/projects/spring-kafka[Spring for Apache Kafka] etc. For many other non-reactive channel adapters thread pools are recommended to avoid blocking during reactive stream processing. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 57e885df82..c7347d1279 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -21,11 +21,23 @@ If you are interested in more details, see the Issue Tracker tickets that were r The `IntegrationPattern` abstraction has been introduced to indicate which enterprise integration pattern (an `IntegrationPatternType`) and category a Spring Integration component belongs to. See its JavaDocs and <<./graph.adoc#integration-graph,Integration Graph>> for more information about this abstraction and its use-cases. +[[x5.3-reactive-message-handler]] +==== `ReactiveMessageHandler` + +The `ReactiveMessageHandler` is now natively supported in the framework. +See <<./reactive-streams.adoc/reactive-message-handler,ReactiveMessageHandler>> for more information. + +[[x5.3-mongodb-reactive-channel-adapters]] +==== MongoDB Reactive Channel Adapters + +`spring-integration-mongodb` module now provides channel adapter implementations for Reactive MongoDB driver support in Spring Data. +See <<./mongodb.adoc#mongodb-reactive-channel-adapters,MongoDB Reactive Channel Adapters>> for more information. + [[x5.3-general]] === General Changes The gateway proxy now doesn't proxy `default` methods by default. -see <<./gateway.adoc/gateway-calling-default-methods,Invoking `default` Methods>> for more information. +See <<./gateway.adoc/gateway-calling-default-methods,Invoking `default` Methods>> for more information. Internal components (such as `_org.springframework.integration.errorLogger`) now have a shortened name when they are represented in the integration graph.