diff --git a/build.gradle b/build.gradle index 4a2ecfa3ff..7d3e475c2a 100644 --- a/build.gradle +++ b/build.gradle @@ -237,6 +237,7 @@ configure(javaProjects) { subproject -> testRuntimeOnly 'org.apache.logging.log4j:log4j-core' testRuntimeOnly 'org.apache.logging.log4j:log4j-jcl' + testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl' } // enable all compiler warnings; individual projects may customize further @@ -471,7 +472,6 @@ project('spring-integration-gemfire') { api "commons-io:commons-io:$commonsIoVersion" testImplementation project(':spring-integration-stream') - testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl' } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 14b9b449a8..ecf37bab9a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.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. @@ -78,14 +78,8 @@ public class FluxMessageChannel extends AbstractMessageChannel Flux.from(publisher) .delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next()) .publishOn(Schedulers.boundedElastic()) - .doOnNext((message) -> { - try { - send(message); - } - catch (Exception e) { - logger.warn("Error during processing event: " + message, e); - } - }) + .doOnNext(this::send) + .onErrorContinue((ex, message) -> logger.warn("Error during processing event: " + message, ex)) .subscribe()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 55b78f0eba..571bd9901f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -42,12 +42,14 @@ import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.endpoint.ReactiveStreamsConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.ChannelResolverUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolver; import org.springframework.scheduling.TaskScheduler; @@ -112,11 +114,17 @@ public class ConsumerEndpointFactoryBean private volatile boolean initialized; - public void setHandler(MessageHandler handler) { - Assert.notNull(handler, "handler must not be null"); + public void setHandler(Object handler) { + Assert.isTrue(handler instanceof MessageHandler || handler instanceof ReactiveMessageHandler, + "'handler' must be an instance of 'MessageHandler' or 'ReactiveMessageHandler'"); synchronized (this.handlerMonitor) { Assert.isNull(this.handler, "handler cannot be overridden"); - this.handler = handler; + if (handler instanceof ReactiveMessageHandler) { + this.handler = new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) handler); + } + else { + this.handler = (MessageHandler) handler; + } } } @@ -210,7 +218,12 @@ public class ConsumerEndpointFactoryBean } } - adviceChain(); + if (!(this.handler instanceof ReactiveMessageHandlerAdapter)) { + adviceChain(); + } + else if (!CollectionUtils.isEmpty(this.adviceChain)) { + LOGGER.warn("the advice chain cannot be applied to a 'ReactiveMessageHandler'"); + } if (this.channelResolver == null) { this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory); } @@ -282,7 +295,13 @@ public class ConsumerEndpointFactoryBean pollingConsumer(channel); } else { - this.endpoint = new ReactiveStreamsConsumer(channel, this.handler); + if (this.handler instanceof ReactiveMessageHandlerAdapter) { + this.endpoint = new ReactiveStreamsConsumer(channel, + ((ReactiveMessageHandlerAdapter) this.handler).getDelegate()); + } + else { + this.endpoint = new ReactiveStreamsConsumer(channel, this.handler); + } } this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java index dc08f25e17..c35e700648 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.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. @@ -22,9 +22,11 @@ import org.springframework.expression.Expression; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.util.StringUtils; /** @@ -47,7 +49,7 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF @Override protected MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { - MessageHandler handler = null; + MessageHandler handler; handler = createDirectHandlerIfPossible(targetObject, targetMethodName); if (handler == null) { handler = configureHandler( @@ -67,26 +69,30 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF */ protected MessageHandler createDirectHandlerIfPossible(final Object targetObject, String targetMethodName) { MessageHandler handler = null; - if (targetObject instanceof MessageHandler - && this.methodIsHandleMessageOrEmpty(targetMethodName)) { + if ((targetObject instanceof MessageHandler || targetObject instanceof ReactiveMessageHandler) + && methodIsHandleMessageOrEmpty(targetMethodName)) { if (targetObject instanceof AbstractMessageProducingHandler) { // should never happen but just return it if it's already an AMPH return (MessageHandler) targetObject; } - /* - * Return a reply-producing message handler so that we still get 'produced no reply' messages - * and the super class will inject the advice chain to advise the handler method if needed. - */ - handler = new ReplyProducingMessageHandlerWrapper((MessageHandler) targetObject); - + if (targetObject instanceof ReactiveMessageHandler) { + handler = new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) targetObject); + } + else { + /* + * Return a reply-producing message handler so that we still get 'produced no reply' messages + * and the super class will inject the advice chain to advise the handler method if needed. + */ + handler = new ReplyProducingMessageHandlerWrapper((MessageHandler) targetObject); + } } return handler; } @Override protected MessageHandler createExpressionEvaluatingHandler(Expression expression) { - ExpressionEvaluatingMessageProcessor processor = new ExpressionEvaluatingMessageProcessor(expression); - processor.setBeanFactory(this.getBeanFactory()); + ExpressionEvaluatingMessageProcessor processor = new ExpressionEvaluatingMessageProcessor<>(expression); + processor.setBeanFactory(getBeanFactory()); ServiceActivatingHandler handler = new ServiceActivatingHandler(processor); handler.setPrimaryExpression(expression); return this.configureHandler(handler); @@ -94,7 +100,7 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF @Override protected MessageHandler createMessageProcessingHandler(MessageProcessor processor) { - return this.configureHandler(new ServiceActivatingHandler(processor)); + return configureHandler(new ServiceActivatingHandler(processor)); } protected MessageHandler configureHandler(ServiceActivatingHandler handler) { @@ -115,7 +121,6 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF @Override protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) { super.postProcessReplyProducer(handler); - if (this.headers != null) { handler.setNotPropagatedHeaders(this.headers); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index 29baf3fd1c..8821462587 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-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. @@ -27,19 +27,25 @@ import org.springframework.integration.channel.ChannelUtils; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.router.MessageRouter; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import reactor.core.CoreSubscriber; import reactor.core.Disposable; -import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; /** + * An {@link AbstractEndpoint} implementation for Reactive Streams subscription into an + * input channel and reactive consumption of messages from that channel. + * * @author Artem Bilan * * @since 5.0 @@ -48,17 +54,22 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra private final MessageChannel inputChannel; - private final MessageHandler handler; - private final Publisher> publisher; + private final MessageHandler handler; + + @Nullable + private final ReactiveMessageHandler reactiveMessageHandler; + + @Nullable private final Subscriber> subscriber; + @Nullable private final Lifecycle lifecycleDelegate; private ErrorHandler errorHandler; - private volatile Subscription subscription; + private volatile Disposable subscription; @SuppressWarnings("unchecked") public ReactiveStreamsConsumer(MessageChannel inputChannel, MessageHandler messageHandler) { @@ -68,10 +79,10 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra : new MessageHandlerSubscriber(messageHandler)); } - public ReactiveStreamsConsumer(MessageChannel inputChannel, final Subscriber> subscriber) { - this.inputChannel = inputChannel; + public ReactiveStreamsConsumer(MessageChannel inputChannel, Subscriber> subscriber) { Assert.notNull(inputChannel, "'inputChannel' must not be null"); Assert.notNull(subscriber, "'subscriber' must not be null"); + this.inputChannel = inputChannel; if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) { logger.warn("The consuming from the NullChannel does not have any effects: " + @@ -90,6 +101,24 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra else { this.handler = this.subscriber::onNext; } + this.reactiveMessageHandler = null; + } + + /** + * Instantiate an endpoint based on the provided {@link MessageChannel} and {@link ReactiveMessageHandler}. + * @param inputChannel the channel to consume in reactive manner. + * @param reactiveMessageHandler the {@link ReactiveMessageHandler} to process messages. + * @since 5.3 + */ + public ReactiveStreamsConsumer(MessageChannel inputChannel, ReactiveMessageHandler reactiveMessageHandler) { + Assert.notNull(inputChannel, "'inputChannel' must not be null"); + this.inputChannel = inputChannel; + this.handler = new ReactiveMessageHandlerAdapter(reactiveMessageHandler); + this.reactiveMessageHandler = reactiveMessageHandler; + this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel); + this.subscriber = null; + this.lifecycleDelegate = + reactiveMessageHandler instanceof Lifecycle ? (Lifecycle) reactiveMessageHandler : null; } public void setErrorHandler(ErrorHandler errorHandler) { @@ -132,55 +161,35 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra if (this.lifecycleDelegate != null) { this.lifecycleDelegate.start(); } - this.publisher.subscribe(new DelegatingSubscriber()); + + Flux flux = null; + if (this.reactiveMessageHandler != null) { + flux = Flux.from(this.publisher) + .flatMap(this.reactiveMessageHandler::handleMessage); + } + else if (this.subscriber != null) { + flux = Flux.from(this.publisher) + .doOnSubscribe(this.subscriber::onSubscribe) + .doOnComplete(this.subscriber::onComplete) + .doOnNext(this.subscriber::onNext); + } + if (flux != null) { + this.subscription = + flux.onErrorContinue((ex, data) -> this.errorHandler.handleError(ex)) + .subscribe(); + } } @Override protected void doStop() { if (this.subscription != null) { - this.subscription.cancel(); + this.subscription.dispose(); } if (this.lifecycleDelegate != null) { this.lifecycleDelegate.stop(); } } - private final class DelegatingSubscriber extends BaseSubscriber> { - - private final Subscriber> delegate = ReactiveStreamsConsumer.this.subscriber; - - DelegatingSubscriber() { - } - - @Override - public void hookOnSubscribe(Subscription s) { - ReactiveStreamsConsumer.this.subscription = s; - this.delegate.onSubscribe(s); - } - - @Override - public void hookOnNext(Message message) { - try { - this.delegate.onNext(message); - } - catch (Exception e) { - ReactiveStreamsConsumer.this.errorHandler.handleError(e); - hookOnError(e); - } - } - - @Override - public void hookOnError(Throwable t) { - this.delegate.onError(t); - } - - @Override - public void hookOnComplete() { - this.delegate.onComplete(); - } - - } - private static final class MessageHandlerSubscriber implements CoreSubscriber>, Disposable, Lifecycle { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ReactiveMessageHandlerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReactiveMessageHandlerAdapter.java new file mode 100644 index 0000000000..31022ce2ee --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReactiveMessageHandlerAdapter.java @@ -0,0 +1,74 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.ReactiveMessageHandler; +import org.springframework.util.Assert; + +/** + * A {@link MessageHandler} implementation to adapt a {@link ReactiveMessageHandler} + * for synchronous invocations. + * A subscription to the returned reactive type from {@link ReactiveMessageHandler#handleMessage(Message)} + * call is done directly in the {@link #handleMessage} implementation. + *

+ * The framework wraps a target {@link ReactiveMessageHandler} into this instance automatically + * for XML and Annotation configuration. For Java DSL it is recommended to wrap for generic usage + * ({@code .handle(MessageHandle)}) or it has to be done in the + * {@link org.springframework.integration.dsl.MessageHandlerSpec} + * implementation for protocol-specif {@link ReactiveMessageHandler}. + *

+ * The framework unwraps a delegate {@link ReactiveMessageHandler} whenever it can compose + * reactive streams, e.g. {@link org.springframework.integration.endpoint.ReactiveStreamsConsumer}. + * + * @author Artem Bilan + * + * @since 5.3 + * + * @see org.springframework.integration.endpoint.ReactiveStreamsConsumer + */ +public class ReactiveMessageHandlerAdapter implements MessageHandler { + + private final ReactiveMessageHandler delegate; + + /** + * Instantiate based on the provided {@link ReactiveMessageHandler}. + * @param reactiveMessageHandler the {@link ReactiveMessageHandler} to delegate to. + */ + public ReactiveMessageHandlerAdapter(ReactiveMessageHandler reactiveMessageHandler) { + Assert.notNull(reactiveMessageHandler, "'reactiveMessageHandler' must not be null"); + this.delegate = reactiveMessageHandler; + } + + /** + * Get access to the delegate {@link ReactiveMessageHandler}. + * Typically used in the framework internally in components which can compose reactive streams internally + * and allow us to avoid an explicit {@code subscribe()} call. + * @return the {@link ReactiveMessageHandler} this instance is delegating to. + */ + public ReactiveMessageHandler getDelegate() { + return this.delegate; + } + + @Override + public void handleMessage(Message message) throws MessagingException { + this.delegate.handleMessage(message).subscribe(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index 27f30d1fab..ece017ee5a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-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. @@ -32,7 +32,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.reactivestreams.Subscriber; @@ -49,8 +49,13 @@ import org.springframework.integration.handler.MethodInvokingMessageHandler; import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.messaging.support.GenericMessage; +import reactor.core.publisher.EmitterProcessor; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + /** * @author Artem Bilan * @@ -261,4 +266,43 @@ public class ReactiveStreamsConsumerTests { assertThat(result).containsExactly(testMessage, testMessage2, testMessage2); } + @Test + public void testReactiveStreamsConsumerFluxMessageChannelReactiveMessageHandler() { + FluxMessageChannel testChannel = new FluxMessageChannel(); + + EmitterProcessor> processor = EmitterProcessor.create(2, false); + + ReactiveMessageHandler messageHandler = + m -> { + processor.onNext(m); + return Mono.empty(); + }; + + ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, messageHandler); + reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); + reactiveConsumer.afterPropertiesSet(); + reactiveConsumer.start(); + + Message testMessage = new GenericMessage<>("test"); + testChannel.send(testMessage); + + reactiveConsumer.stop(); + + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> testChannel.send(testMessage)) + .withCauseInstanceOf(IllegalStateException.class) + .withMessageContaining("doesn't have subscribers to accept messages"); + + reactiveConsumer.start(); + + Message testMessage2 = new GenericMessage<>("test2"); + testChannel.send(testMessage2); + + StepVerifier.create(processor) + .expectNext(testMessage, testMessage2) + .thenCancel() + .verify(); + } + + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index 0c900c722a..6e6fcabfa0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-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. @@ -18,7 +18,6 @@ package org.springframework.integration.dsl.flows; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import static org.assertj.core.api.Assertions.fail; import java.io.Serializable; import java.util.concurrent.CountDownLatch; @@ -33,8 +32,7 @@ import java.util.function.Supplier; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.ListableBeanFactory; @@ -64,8 +62,8 @@ import org.springframework.integration.dsl.Pollers; import org.springframework.integration.dsl.Transformers; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.integration.handler.GenericHandler; import org.springframework.integration.handler.LoggingHandler; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice; @@ -93,7 +91,9 @@ import org.springframework.stereotype.Component; import org.springframework.stereotype.Service; 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; /** * @author Artem Bilan @@ -104,7 +104,7 @@ import org.springframework.test.context.junit4.SpringRunner; * @since 5.0 */ @ContextConfiguration(loader = NoBeansOverrideAnnotationConfigContextLoader.class) -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class IntegrationFlowTests { @@ -191,15 +191,12 @@ public class IntegrationFlowTests { assertThat(this.beanFactory.containsBean("expressionFilter.handler")).isTrue(); QueueChannel replyChannel = new QueueChannel(); Message message = MessageBuilder.withPayload("100").setReplyChannel(replyChannel).build(); - try { - this.inputChannel.send(message); - fail("Expected MessageDispatchingException"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageDeliveryException.class); - assertThat(e.getCause()).isInstanceOf(MessageDispatchingException.class); - assertThat(e.getMessage()).contains("Dispatcher has no subscribers"); - } + + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> this.inputChannel.send(message)) + .withCauseInstanceOf(MessageDispatchingException.class) + .withMessageContaining("Dispatcher has no subscribers"); + this.controlBus.send("@payloadSerializingTransformer.start()"); final AtomicBoolean used = new AtomicBoolean(); @@ -670,7 +667,7 @@ public class IntegrationFlowTests { public IntegrationFlow wireTapFlow1() { return IntegrationFlows.from("tappedChannel1") .wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo"))) - .handle(loggingMessageHandler()) + .handle(new ReactiveMessageHandlerAdapter((message) -> Mono.just(message).log().then())) .get(); } @@ -820,9 +817,10 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow errorRecovererFlow() { return IntegrationFlows.from(Function.class, (gateway) -> gateway.beanName("errorRecovererFunction")) - .handle((GenericHandler) (p, h) -> { - throw new RuntimeException("intentional"); - }, e -> e.advice(retryAdvice())) + .handle((p, h) -> { + throw new RuntimeException("intentional"); + }, + e -> e.advice(retryAdvice())) .get(); } @@ -903,7 +901,7 @@ public class IntegrationFlowTests { return IntegrationFlows.from(Consumer.class, (gateway) -> gateway.beanName("globalErrorChannelResolutionFunction")) .channel(c -> c.executor(taskExecutor)) - .handle((GenericHandler) (p, h) -> { + .handle((p, h) -> { throw new RuntimeException("intentional"); }) .get(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdviceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdviceTests.java index efd5488180..7e567acbf6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdviceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdviceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-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. @@ -19,8 +19,7 @@ package org.springframework.integration.handler.advice; import static org.assertj.core.api.Assertions.assertThat; import org.aopalliance.aop.Advice; -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; @@ -28,21 +27,22 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.handler.GenericHandler; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice.MessageHandlingExpressionEvaluatingAdviceException; import org.springframework.integration.message.AdviceMessage; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell + * @author Artem Bilan + * * @since 5.0 * */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig public class ExpressionEvaluatingRequestHandlerAdviceTests { @Autowired @@ -70,14 +70,16 @@ public class ExpressionEvaluatingRequestHandlerAdviceTests { @Bean public IntegrationFlow advised() { - return f -> f.handle((GenericHandler) (payload, headers) -> { - if (payload.equals("good")) { - return null; - } - else { - throw new RuntimeException("some failure"); - } - }, c -> c.advice(expressionAdvice())); + return f -> f + .handle((payload, headers) -> { + if (payload.equals("good")) { + return null; + } + else { + throw new RuntimeException("some failure"); + } + }, + c -> c.advice(expressionAdvice())); } @Bean diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java index f5bc111b5e..fec15feea0 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java @@ -17,7 +17,9 @@ package org.springframework.integration.mongodb.dsl; import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; import org.springframework.data.mongodb.core.MongoOperations; +import org.springframework.data.mongodb.core.ReactiveMongoOperations; import org.springframework.data.mongodb.core.convert.MongoConverter; /** @@ -52,6 +54,30 @@ public final class MongoDb { return new MongoDbOutboundGatewaySpec(mongoTemplate); } + /** + * Create a {@link ReactiveMongoDbMessageHandlerSpec} builder instance + * based on the provided {@link ReactiveMongoDatabaseFactory}. + * @param mongoDbFactory the {@link ReactiveMongoDatabaseFactory} to use. + * @return the {@link MongoDbOutboundGatewaySpec} instance + * @since 5.3 + */ + public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter( + ReactiveMongoDatabaseFactory mongoDbFactory) { + + return new ReactiveMongoDbMessageHandlerSpec(mongoDbFactory); + } + + /** + * Create a {@link ReactiveMongoDbMessageHandlerSpec} builder instance + * based on the provided {@link ReactiveMongoOperations}. + * @param mongoTemplate the {@link ReactiveMongoOperations} to use. + * @return the {@link ReactiveMongoDbMessageHandlerSpec} instance + * @since 5.3 + */ + public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter(ReactiveMongoOperations mongoTemplate) { + return new ReactiveMongoDbMessageHandlerSpec(mongoTemplate); + } + private MongoDb() { } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbOutboundGatewaySpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbOutboundGatewaySpec.java index 41b2061157..25b5ae4f45 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbOutboundGatewaySpec.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbOutboundGatewaySpec.java @@ -32,7 +32,7 @@ import org.springframework.messaging.Message; /** * A {@link MessageHandlerSpec} extension for the MongoDb Outbound endpoint {@link MongoDbOutboundGateway} * - * @author Xavier Padr? + * @author Xavier Padro * @author Artem Bilan * * @since 5.0 diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java new file mode 100644 index 0000000000..f5fbb5dda8 --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java @@ -0,0 +1,108 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.mongodb.dsl; + +import java.util.Collections; +import java.util.Map; +import java.util.function.Function; + +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.ReactiveMongoOperations; +import org.springframework.data.mongodb.core.convert.MongoConverter; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.dsl.ComponentsRegistration; +import org.springframework.integration.dsl.MessageHandlerSpec; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; +import org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandler; +import org.springframework.messaging.Message; + +/** + * A {@link MessageHandlerSpec} extension for the Reactive MongoDb Outbound endpoint + * {@link ReactiveMongoDbStoringMessageHandler}. + * + * @author Artem Bilan + * + * @since 5.3 + */ +public class ReactiveMongoDbMessageHandlerSpec + extends MessageHandlerSpec + implements ComponentsRegistration { + + private final ReactiveMongoDbStoringMessageHandler messageHandler; + + ReactiveMongoDbMessageHandlerSpec(ReactiveMongoDatabaseFactory mongoDbFactory) { + this(new ReactiveMongoDbStoringMessageHandler(mongoDbFactory)); + } + + ReactiveMongoDbMessageHandlerSpec(ReactiveMongoOperations reactiveMongoOperations) { + this(new ReactiveMongoDbStoringMessageHandler(reactiveMongoOperations)); + } + + private ReactiveMongoDbMessageHandlerSpec(ReactiveMongoDbStoringMessageHandler messageHandler) { + this.messageHandler = messageHandler; + this.target = new ReactiveMessageHandlerAdapter(this.messageHandler); + } + + /** + * Configure a {@link MongoConverter}. + * @param mongoConverter the {@link MongoConverter} to use. + * @return the spec + */ + public ReactiveMongoDbMessageHandlerSpec mongoConverter(MongoConverter mongoConverter) { + this.messageHandler.setMongoConverter(mongoConverter); + return this; + } + + /** + * Configure a collection name to store data. + * @param collectionName the explicit collection name to use. + * @return the spec + */ + public ReactiveMongoDbMessageHandlerSpec collectionName(String collectionName) { + return collectionNameExpression(new LiteralExpression(collectionName)); + } + + /** + * Configure a {@link Function} for evaluation a collection against request message. + * @param collectionNameFunction the {@link Function} to determine a collection name at runtime. + * @param

an expected payload type + * @return the spec + */ + public

ReactiveMongoDbMessageHandlerSpec collectionNameFunction( + Function, String> collectionNameFunction) { + + return collectionNameExpression(new FunctionExpression<>(collectionNameFunction)); + } + + /** + * Configure a SpEL expression to evaluate a collection name against a request message. + * @param collectionNameExpression the SpEL expression to use. + * @return the spec + */ + public ReactiveMongoDbMessageHandlerSpec collectionNameExpression(Expression collectionNameExpression) { + this.messageHandler.setCollectionNameExpression(collectionNameExpression); + return this; + } + + @Override + public Map getComponentsToRegister() { + return Collections.singletonMap(this.messageHandler, null); + } + +} diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandler.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandler.java index 389a17fd8d..dc9603e10d 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandler.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-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. @@ -36,6 +36,7 @@ import reactor.core.publisher.Mono; * collection is identified by evaluation of the {@link #collectionNameExpression}. * * @author David Turanski + * @author Artme Bilan * * @since 5.3 */ @@ -104,7 +105,9 @@ public class ReactiveMongoDbStoringMessageHandler extends AbstractReactiveMessag super.onInit(); this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); if (this.mongoTemplate == null) { - this.mongoTemplate = new ReactiveMongoTemplate(this.mongoDbFactory, this.mongoConverter); + ReactiveMongoTemplate mongoTemplate = new ReactiveMongoTemplate(this.mongoDbFactory, this.mongoConverter); + mongoTemplate.setApplicationContext(getApplicationContext()); + this.mongoTemplate = mongoTemplate; } this.initialized = true; } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java index 7d9cd67dc8..edfa474d73 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java @@ -17,7 +17,9 @@ package org.springframework.integration.mongodb.dsl; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import java.time.Duration; import java.util.Arrays; import java.util.List; @@ -34,8 +36,10 @@ import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.BulkOperations; import org.springframework.data.mongodb.core.MongoOperations; import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.data.mongodb.core.convert.MongoConverter; import org.springframework.data.mongodb.core.mapping.MongoMappingContext; +import org.springframework.data.mongodb.core.query.BasicQuery; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.integration.config.EnableIntegration; @@ -255,6 +259,25 @@ public class MongoDbTests extends MongoDbAvailableTests { bulkOperations.execute(); } + @Autowired + @Qualifier("reactiveStore.input") + private MessageChannel reactiveStoreInput; + + @Test + @MongoDbAvailable + public void testReactiveMongoDbMessageHandler() { + this.reactiveStoreInput.send(MessageBuilder.withPayload(createPerson("Bob")).build()); + + ReactiveMongoTemplate reactiveMongoTemplate = new ReactiveMongoTemplate(REACTIVE_MONGO_DATABASE_FACTORY); + + await().untilAsserted(() -> + assertThat( + reactiveMongoTemplate.findOne(new BasicQuery("{'name' : 'Bob'}"), Person.class, "data") + .block(Duration.ofSeconds(10))) + .isNotNull() + .extracting("name", "address.state").contains("Bob", "PA")); + } + @Configuration @EnableIntegration public static class ContextConfiguration { @@ -395,6 +418,14 @@ public class MongoDbTests extends MongoDbAvailableTests { .entityClass(Person.class); } + @Bean + public IntegrationFlow reactiveStore() { + return f -> f + .channel(MessageChannels.flux()) + .handle(MongoDb.reactiveOutboundChannelAdapter(REACTIVE_MONGO_DATABASE_FACTORY)); + } + + } } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests.java index 0646859c3f..f30baf55be 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests.java @@ -39,7 +39,9 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; /** - * @author Xavier Padr? + * @author Xavier Padro + * @author Artem Bilan + * * @since 5.0 */ @RunWith(SpringRunner.class) @@ -52,7 +54,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { private ApplicationContext context; @Before - public void setUp() throws Exception { + public void setUp() { MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); MongoTemplate mongoTemplate = new MongoTemplate(mongoDbFactory); @@ -63,7 +65,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { } @After - public void cleanUp() throws Exception { + public void cleanUp() { MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); MongoTemplate mongoTemplate = new MongoTemplate(mongoDbFactory); @@ -73,7 +75,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void testSingleQuery() throws Exception { + public void testSingleQuery() { EventDrivenConsumer consumer = context.getBean("gatewaySingleQuery", EventDrivenConsumer.class); PollableChannel outChannel = context.getBean("out", PollableChannel.class); @@ -87,7 +89,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void testSingleQueryWithTemplate() throws Exception { + public void testSingleQueryWithTemplate() { EventDrivenConsumer consumer = context.getBean("gatewayWithTemplate", EventDrivenConsumer.class); PollableChannel outChannel = context.getBean("out", PollableChannel.class); @@ -101,7 +103,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void testSingleQueryExpression() throws Exception { + public void testSingleQueryExpression() { EventDrivenConsumer consumer = context.getBean("gatewaySingleQueryExpression", EventDrivenConsumer.class); PollableChannel outChannel = context.getBean("out", PollableChannel.class); @@ -120,7 +122,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void testQueryExpression() throws Exception { + public void testQueryExpression() { EventDrivenConsumer consumer = context.getBean("gatewayQueryExpression", EventDrivenConsumer.class); PollableChannel outChannel = context.getBean("out", PollableChannel.class); @@ -139,7 +141,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void testQueryExpressionWithLimit() throws Exception { + public void testQueryExpressionWithLimit() { EventDrivenConsumer consumer = context.getBean("gatewayQueryExpressionLimit", EventDrivenConsumer.class); PollableChannel outChannel = context.getBean("out", PollableChannel.class); @@ -157,7 +159,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void testCollectionCallback() throws Exception { + public void testCollectionCallback() { EventDrivenConsumer consumer = context.getBean("gatewayCollectionCallback", EventDrivenConsumer.class); PollableChannel outChannel = context.getBean("out", PollableChannel.class); diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests-context.xml new file mode 100644 index 0000000000..796319486a --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests.java index 29941931cd..7de1e493e1 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-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. @@ -18,6 +18,7 @@ package org.springframework.integration.mongodb.outbound; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.awaitility.Awaitility.await; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; @@ -26,8 +27,12 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Before; import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Answers; import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; import org.springframework.data.mongodb.core.ReactiveMongoOperations; import org.springframework.data.mongodb.core.ReactiveMongoTemplate; @@ -40,6 +45,9 @@ import org.springframework.integration.mongodb.rules.MongoDbAvailable; import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; import reactor.core.publisher.Mono; @@ -48,19 +56,25 @@ import reactor.core.publisher.Mono; * @author Oleg Zhurakousky * @author Gary Russell * @author David Turanski + * @author Artem Bilan * * @since 5.3 */ +@RunWith(SpringRunner.class) +@DirtiesContext public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { private ReactiveMongoTemplate template; private ReactiveMongoDatabaseFactory mongoDbFactory; + @Autowired + private MessageChannel input; + @Before public void setUp() { - mongoDbFactory = this.prepareReactiveMongoFactory("foo"); - template = new ReactiveMongoTemplate(mongoDbFactory); + this.mongoDbFactory = prepareReactiveMongoFactory("foo"); + this.template = new ReactiveMongoTemplate(this.mongoDbFactory); } @Test @@ -82,6 +96,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT public void validateMessageHandlingWithDefaultCollection() { ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); waitFor(handler.handleMessage(message)); @@ -99,6 +114,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); handler.setCollectionNameExpression(new LiteralExpression("foo")); handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); @@ -119,6 +135,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); handler.setCollectionNameExpression(new LiteralExpression(null)); handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(createPerson("Bob")).build(); @@ -139,6 +156,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT converter = spy(converter); handler.setMongoConverter(converter); handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); waitFor(handler.handleMessage(message)); @@ -161,6 +179,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(writingTemplate); handler.setCollectionNameExpression(new LiteralExpression("foo")); handler.setBeanFactory(mock(BeanFactory.class)); + handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); waitFor(handler.handleMessage(message)); @@ -172,8 +191,22 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT assertThat(person.getAddress().getState()).isEqualTo("PA"); } + @Test + @MongoDbAvailable + public void testReactiveMongoMessageHandlerFromApplicationContext() { + Message message = MessageBuilder.withPayload(createPerson("Bob")).build(); + this.input.send(message); + + Query query = new BasicQuery("{'name' : 'Bob'}"); + + await().untilAsserted(() -> + assertThat(waitFor(this.template.findOne(query, Person.class, "data"))) + .isNotNull() + .extracting("name", "address.state").contains("Bob", "PA")); + } + private static T waitFor(Mono mono) { - return mono.block(Duration.ofSeconds(3)); + return mono.block(Duration.ofSeconds(10)); } } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java index be477c0117..bd143a2655 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java @@ -65,16 +65,20 @@ public abstract class MongoDbAvailableTests { MongoClientSettings.builder().build()), "test"); + public static final ReactiveMongoDatabaseFactory REACTIVE_MONGO_DATABASE_FACTORY = + new SimpleReactiveMongoDatabaseFactory( + com.mongodb.reactivestreams.client.MongoClients.create( + MongoClientSettings.builder().build()), + "test"); + protected MongoDbFactory prepareMongoFactory(String... additionalCollectionsToDrop) { cleanupCollections(MONGO_DATABASE_FACTORY, additionalCollectionsToDrop); return MONGO_DATABASE_FACTORY; } protected ReactiveMongoDatabaseFactory prepareReactiveMongoFactory(String... additionalCollectionsToDrop) { - ReactiveMongoDatabaseFactory mongoDbFactory = new SimpleReactiveMongoDatabaseFactory( - com.mongodb.reactivestreams.client.MongoClients.create(), "test"); - cleanupCollections(mongoDbFactory, additionalCollectionsToDrop); - return mongoDbFactory; + cleanupCollections(REACTIVE_MONGO_DATABASE_FACTORY, additionalCollectionsToDrop); + return REACTIVE_MONGO_DATABASE_FACTORY; } protected void cleanupCollections(ReactiveMongoDatabaseFactory mongoDbFactory,