diff --git a/build.gradle b/build.gradle index c9fd8df3db..1988d52803 100644 --- a/build.gradle +++ b/build.gradle @@ -80,6 +80,7 @@ ext { log4jVersion = '2.12.1' micrometerVersion = '1.3.2' mockitoVersion = '3.2.0' + mongodbReactiveDriverVersion = '1.12.0' mysqlVersion = '8.0.18' pahoMqttClientVersion = '1.2.0' postgresVersion = '42.2.8' @@ -574,6 +575,8 @@ project('spring-integration-mongodb') { compile('org.springframework.data:spring-data-mongodb') { exclude group: 'org.springframework' } + compile("org.mongodb:mongodb-driver-reactivestreams:$mongodbReactiveDriverVersion", optional) + testCompile 'io.projectreactor:reactor-test' } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 7338020e02..e2530212f9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -16,28 +16,13 @@ package org.springframework.integration.handler; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - import org.reactivestreams.Subscription; -import org.springframework.core.Ordered; -import org.springframework.integration.IntegrationPattern; -import org.springframework.integration.IntegrationPatternType; -import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.context.Orderable; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.support.management.AbstractMessageHandlerMetrics; -import org.springframework.integration.support.management.ConfigurableMetricsAware; -import org.springframework.integration.support.management.DefaultMessageHandlerMetrics; -import org.springframework.integration.support.management.IntegrationManagedResource; -import org.springframework.integration.support.management.TrackableComponent; -import org.springframework.integration.support.management.metrics.MeterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.support.management.metrics.SampleFacade; -import org.springframework.integration.support.management.metrics.TimerFacade; import org.springframework.integration.support.utils.IntegrationUtils; -import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -45,141 +30,38 @@ import org.springframework.util.Assert; import reactor.core.CoreSubscriber; /** - * Base class for MessageHandler implementations that provides basic validation - * and error handling capabilities. Asserts that the incoming Message is not - * null and that it does not contain a null payload. Converts checked exceptions - * into runtime {@link org.springframework.messaging.MessagingException}s. + * Base class for {@link MessageHandler} implementations. * - * @author Mark Fisher - * @author Oleg Zhurakousky - * @author Gary Russell + * @author David Turanski * @author Artem Bilan - * @author Amit Sadafule */ -@SuppressWarnings("deprecation") -@IntegrationManagedResource -public abstract class AbstractMessageHandler extends IntegrationObjectSupport - implements MessageHandler, - org.springframework.integration.support.management.MessageHandlerMetrics, - ConfigurableMetricsAware, - TrackableComponent, Orderable, CoreSubscriber>, - IntegrationPattern { +public abstract class AbstractMessageHandler extends MessageHandlerSupport + implements MessageHandler, CoreSubscriber> { - private final ManagementOverrides managementOverrides = new ManagementOverrides(); - - private final Set timers = ConcurrentHashMap.newKeySet(); - - private volatile boolean shouldTrack = false; - - private volatile int order = Ordered.LOWEST_PRECEDENCE; - - private volatile AbstractMessageHandlerMetrics handlerMetrics = new DefaultMessageHandlerMetrics(); - - private volatile boolean statsEnabled; - - private volatile boolean countsEnabled; - - private volatile String managedName; - - private volatile String managedType; - - private volatile boolean loggingEnabled = true; - - private MetricsCaptor metricsCaptor; - - private TimerFacade successTimer; - - @Override - public boolean isLoggingEnabled() { - return this.loggingEnabled; - } - - @Override - public void setLoggingEnabled(boolean loggingEnabled) { - this.loggingEnabled = loggingEnabled; - this.managementOverrides.loggingConfigured = true; - } - - @Override - public void registerMetricsCaptor(MetricsCaptor metricsCaptorToRegister) { - this.metricsCaptor = metricsCaptorToRegister; - } - - @Nullable - protected MetricsCaptor getMetricsCaptor() { - return this.metricsCaptor; - } - - @Override - public void setOrder(int order) { - this.order = order; - } - - @Override - public int getOrder() { - return this.order; - } - - @Override - public String getComponentType() { - return "message-handler"; - } - - @Override - public void setShouldTrack(boolean shouldTrack) { - this.shouldTrack = shouldTrack; - } - - @Override - public void configureMetrics(AbstractMessageHandlerMetrics metrics) { - Assert.notNull(metrics, "'metrics' must not be null"); - this.handlerMetrics = metrics; - this.managementOverrides.metricsConfigured = true; - } - - @Override - public ManagementOverrides getOverrides() { - return this.managementOverrides; - } - - @Override - public IntegrationPatternType getIntegrationPatternType() { - return IntegrationPatternType.outbound_channel_adapter; - } - - @Override - protected void onInit() { - if (this.statsEnabled) { - this.handlerMetrics.setFullStatsEnabled(true); - } - } - - @Override - public void handleMessage(Message messageArg) { - Message message = messageArg; + @SuppressWarnings("deprecation") + public void handleMessage(Message message) { Assert.notNull(message, "Message must not be null"); - Assert.notNull(message.getPayload(), "Message payload must not be null"); //NOSONAR - false positive - if (this.loggingEnabled && this.logger.isDebugEnabled()) { + if (isLoggingEnabled() && this.logger.isDebugEnabled()) { this.logger.debug(this + " received message: " + message); } org.springframework.integration.support.management.MetricsContext start = null; - boolean countsAreEnabled = this.countsEnabled; - AbstractMessageHandlerMetrics metrics = this.handlerMetrics; SampleFacade sample = null; - if (countsAreEnabled && this.metricsCaptor != null) { - sample = this.metricsCaptor.start(); + MetricsCaptor metricsCaptor = getMetricsCaptor(); + if (metricsCaptor != null && isCountsEnabled()) { + sample = metricsCaptor.start(); } try { - if (this.shouldTrack) { + if (shouldTrack()) { message = MessageHistory.write(message, this, getMessageBuilderFactory()); } - if (countsAreEnabled) { - start = metrics.beforeHandle(); + AbstractMessageHandlerMetrics handlerMetrics = getHandlerMetrics(); + if (isCountsEnabled()) { + start = handlerMetrics.beforeHandle(); handleMessageInternal(message); if (sample != null) { sample.stop(sendTimer()); } - metrics.afterHandle(start, true); + handlerMetrics.afterHandle(start, true); } else { handleMessageInternal(message); @@ -189,44 +71,20 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport if (sample != null) { sample.stop(buildSendTimer(false, e.getClass().getSimpleName())); } - if (countsAreEnabled) { - metrics.afterHandle(start, false); + if (isCountsEnabled()) { + getHandlerMetrics().afterHandle(start, false); } throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, () -> "error occurred in message handler [" + this + "]", e); } } - private TimerFacade sendTimer() { - if (this.successTimer == null) { - this.successTimer = buildSendTimer(true, "none"); - } - return this.successTimer; - } - - private TimerFacade buildSendTimer(boolean success, String exception) { - TimerFacade timer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME) - .tag("type", "handler") - .tag("name", getComponentName() == null ? "unknown" : getComponentName()) - .tag("result", success ? "success" : "failure") - .tag("exception", exception) - .description("Send processing time") - .build(); - this.timers.add(timer); - return timer; - } - @Override public void onSubscribe(Subscription subscription) { Assert.notNull(subscription, "'subscription' must not be null"); subscription.request(Long.MAX_VALUE); } - @Override - public void onNext(Message message) { - handleMessage(message); - } - @Override public void onError(Throwable throwable) { @@ -237,125 +95,11 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport } + @Override + public void onNext(Message message) { + handleMessage(message); + } + protected abstract void handleMessageInternal(Message message); - @Override - public void reset() { - this.handlerMetrics.reset(); - } - - @Override - public long getHandleCountLong() { - return this.handlerMetrics.getHandleCountLong(); - } - - @Override - public int getHandleCount() { - return this.handlerMetrics.getHandleCount(); - } - - @Override - public int getErrorCount() { - return this.handlerMetrics.getErrorCount(); - } - - @Override - public long getErrorCountLong() { - return this.handlerMetrics.getErrorCountLong(); - } - - @Override - public double getMeanDuration() { - return this.handlerMetrics.getMeanDuration(); - } - - @Override - public double getMinDuration() { - return this.handlerMetrics.getMinDuration(); - } - - @Override - public double getMaxDuration() { - return this.handlerMetrics.getMaxDuration(); - } - - @Override - public double getStandardDeviationDuration() { - return this.handlerMetrics.getStandardDeviationDuration(); - } - - @Override - public int getActiveCount() { - return this.handlerMetrics.getActiveCount(); - } - - @Override - public long getActiveCountLong() { - return this.handlerMetrics.getActiveCountLong(); - } - - @Override - public org.springframework.integration.support.management.Statistics getDuration() { - return this.handlerMetrics.getDuration(); - } - - @Override - public void setStatsEnabled(boolean statsEnabled) { - if (statsEnabled) { - this.countsEnabled = true; - this.managementOverrides.countsConfigured = true; - } - this.statsEnabled = statsEnabled; - if (this.handlerMetrics != null) { - this.handlerMetrics.setFullStatsEnabled(statsEnabled); - } - this.managementOverrides.statsConfigured = true; - } - - @Override - public boolean isStatsEnabled() { - return this.statsEnabled; - } - - @Override - public void setCountsEnabled(boolean countsEnabled) { - this.countsEnabled = countsEnabled; - this.managementOverrides.countsConfigured = true; - if (!countsEnabled) { - this.statsEnabled = false; - this.managementOverrides.statsConfigured = true; - } - } - - @Override - public boolean isCountsEnabled() { - return this.countsEnabled; - } - - @Override - public void setManagedName(String managedName) { - this.managedName = managedName; - } - - @Override - public String getManagedName() { - return this.managedName; - } - - @Override - public void setManagedType(String managedType) { - this.managedType = managedType; - } - - @Override - public String getManagedType() { - return this.managedType; - } - - @Override - public void destroy() { - this.timers.forEach(MeterFacade::remove); - this.timers.clear(); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReactiveMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReactiveMessageHandler.java new file mode 100644 index 0000000000..495dda14cf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReactiveMessageHandler.java @@ -0,0 +1,55 @@ +/* + * Copyright 2019 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.integration.history.MessageHistory; +import org.springframework.messaging.Message; +import org.springframework.messaging.ReactiveMessageHandler; +import org.springframework.util.Assert; + +import reactor.core.publisher.Mono; + +/** + * Base class for {@link ReactiveMessageHandler} implementations. + * + * @author David Turanski + * @author Artem Bilan + * + * @since 5.3 + */ +public abstract class AbstractReactiveMessageHandler extends MessageHandlerSupport + implements ReactiveMessageHandler { + + @Override + public Mono handleMessage(Message message) { + Assert.notNull(message, "Message must not be null"); + if (isLoggingEnabled() && this.logger.isDebugEnabled()) { + this.logger.debug(this + " received message: " + message); + } + + if (shouldTrack()) { + message = MessageHistory.write(message, this, getMessageBuilderFactory()); + } + final Message msg = message; + return handleMessageInternal(msg) + .doOnError(e -> this.logger.error( + "An error occurred in message handler [" + this + "] on message [" + msg + "]", e)); + } + + protected abstract Mono handleMessageInternal(Message message); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java new file mode 100644 index 0000000000..741f9918bd --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerSupport.java @@ -0,0 +1,294 @@ +/* + * Copyright 2019 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 java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.core.Ordered; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; +import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.integration.context.Orderable; +import org.springframework.integration.support.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.support.management.ConfigurableMetricsAware; +import org.springframework.integration.support.management.DefaultMessageHandlerMetrics; +import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.management.TrackableComponent; +import org.springframework.integration.support.management.metrics.MeterFacade; +import org.springframework.integration.support.management.metrics.MetricsCaptor; +import org.springframework.integration.support.management.metrics.TimerFacade; +import org.springframework.util.Assert; + +/** + * Base class for Message handling components that provides basic validation and error + * handling capabilities. Asserts that the incoming Message is not null and that it does + * not contain a null payload. Converts checked exceptions into runtime + * {@link org.springframework.messaging.MessagingException}s. + * + * @author Mark Fisher + * @author Oleg Zhurakousky + * @author Gary Russell + * @author Artem Bilan + * @author Amit Sadafule + * @author David Turanski + * + * @since 5.3 + * + */ +@SuppressWarnings("deprecation") +@IntegrationManagedResource +public abstract class MessageHandlerSupport extends IntegrationObjectSupport + implements org.springframework.integration.support.management.MessageHandlerMetrics, + ConfigurableMetricsAware, + TrackableComponent, Orderable, IntegrationPattern { + + private final ManagementOverrides managementOverrides = new ManagementOverrides(); + + private final Set timers = ConcurrentHashMap.newKeySet(); + + private boolean shouldTrack = false; + + private AbstractMessageHandlerMetrics handlerMetrics = new DefaultMessageHandlerMetrics(); + + private boolean countsEnabled; + + private boolean loggingEnabled = true; + + private MetricsCaptor metricsCaptor; + + private int order = Ordered.LOWEST_PRECEDENCE; + + private boolean statsEnabled; + + private String managedName; + + private String managedType; + + private TimerFacade successTimer; + + @Override + public boolean isLoggingEnabled() { + return this.loggingEnabled; + } + + @Override + public void setLoggingEnabled(boolean loggingEnabled) { + this.loggingEnabled = loggingEnabled; + this.managementOverrides.loggingConfigured = true; + } + + @Override + public void registerMetricsCaptor(MetricsCaptor metricsCaptorToRegister) { + this.metricsCaptor = metricsCaptorToRegister; + } + + protected AbstractMessageHandlerMetrics getHandlerMetrics() { + return this.handlerMetrics; + } + + protected MetricsCaptor getMetricsCaptor() { + return this.metricsCaptor; + } + + @Override + public void setOrder(int order) { + this.order = order; + } + + @Override + public int getOrder() { + return this.order; + } + + @Override + public String getComponentType() { + return "message-handler"; + } + + @Override + public void setShouldTrack(boolean shouldTrack) { + this.shouldTrack = shouldTrack; + } + + protected boolean shouldTrack() { + return this.shouldTrack; + } + + @Override + public void configureMetrics(AbstractMessageHandlerMetrics metrics) { + Assert.notNull(metrics, "'metrics' must not be null"); + this.handlerMetrics = metrics; + this.managementOverrides.metricsConfigured = true; + } + + @Override + public ManagementOverrides getOverrides() { + return this.managementOverrides; + } + + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.outbound_channel_adapter; + } + + @Override + protected void onInit() { + if (this.statsEnabled) { + this.handlerMetrics.setFullStatsEnabled(true); + } + } + + protected TimerFacade sendTimer() { + if (this.successTimer == null) { + this.successTimer = buildSendTimer(true, "none"); + } + return this.successTimer; + } + + protected TimerFacade buildSendTimer(boolean success, String exception) { + TimerFacade timer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME) + .tag("type", "handler") + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("result", success ? "success" : "failure") + .tag("exception", exception) + .description("Send processing time") + .build(); + this.timers.add(timer); + return timer; + } + + @Override + public void reset() { + this.handlerMetrics.reset(); + } + + @Override + public long getHandleCountLong() { + return this.handlerMetrics.getHandleCountLong(); + } + + @Override + public int getHandleCount() { + return this.handlerMetrics.getHandleCount(); + } + + @Override + public int getErrorCount() { + return this.handlerMetrics.getErrorCount(); + } + + @Override + public long getErrorCountLong() { + return this.handlerMetrics.getErrorCountLong(); + } + + @Override + public double getMeanDuration() { + return this.handlerMetrics.getMeanDuration(); + } + + @Override + public double getMinDuration() { + return this.handlerMetrics.getMinDuration(); + } + + @Override + public double getMaxDuration() { + return this.handlerMetrics.getMaxDuration(); + } + + @Override + public double getStandardDeviationDuration() { + return this.handlerMetrics.getStandardDeviationDuration(); + } + + @Override + public int getActiveCount() { + return this.handlerMetrics.getActiveCount(); + } + + @Override + public long getActiveCountLong() { + return this.handlerMetrics.getActiveCountLong(); + } + + @Override + public org.springframework.integration.support.management.Statistics getDuration() { + return this.handlerMetrics.getDuration(); + } + + @Override + public void setStatsEnabled(boolean statsEnabled) { + if (statsEnabled) { + this.countsEnabled = true; + this.managementOverrides.countsConfigured = true; + } + this.statsEnabled = statsEnabled; + if (this.handlerMetrics != null) { + this.handlerMetrics.setFullStatsEnabled(statsEnabled); + } + this.managementOverrides.statsConfigured = true; + } + + @Override + public boolean isStatsEnabled() { + return this.statsEnabled; + } + + @Override + public void setCountsEnabled(boolean countsEnabled) { + this.countsEnabled = countsEnabled; + this.managementOverrides.countsConfigured = true; + if (!countsEnabled) { + this.statsEnabled = false; + this.managementOverrides.statsConfigured = true; + } + } + + @Override + public boolean isCountsEnabled() { + return this.countsEnabled; + } + + @Override + public void setManagedName(String managedName) { + this.managedName = managedName; + } + + @Override + public String getManagedName() { + return this.managedName; + } + + @Override + public void setManagedType(String managedType) { + this.managedType = managedType; + } + + @Override + public String getManagedType() { + return this.managedType; + } + + @Override + public void destroy() { + this.timers.forEach(MeterFacade::remove); + this.timers.clear(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/ReactiveMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/ReactiveMessageHandlerTests.java new file mode 100644 index 0000000000..5d2bd7d2ff --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/ReactiveMessageHandlerTests.java @@ -0,0 +1,90 @@ +/* + * Copyright 2019 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 static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.history.MessageHistory; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +import reactor.core.publisher.Mono; + +/** + * @author David Turanski + * @author Artem Bilan + * + * @since 5.3 + */ +class ReactiveMessageHandlerTests { + + private AtomicBoolean handled = new AtomicBoolean(); + + private QueueChannel output = new QueueChannel(); + + @BeforeEach + void setUp() { + handled.set(false); + } + + @Test + void messageHandledOnSubscribe() { + assertThat(handled.get()).isFalse(); + TestReactiveMessageHandler handler = new TestReactiveMessageHandler(); + handler.afterPropertiesSet(); + Message message = new GenericMessage<>(""); + handler.handleMessage(message).subscribe(); + assertThat(handled.get()).isTrue(); + } + + @Test + void messageTracked() { + assertThat(handled.get()).isFalse(); + TestReactiveMessageHandler handler = new TestReactiveMessageHandler(); + handler.setShouldTrack(true); + handler.setComponentName("test-message-handler"); + handler.afterPropertiesSet(); + Message message = new GenericMessage<>(""); + handler.handleMessage(message).subscribe(); + assertThat(handled.get()).isTrue(); + Message received = output.receive(1000); + assertThat(received).isNotNull(); + MessageHistory history = received.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class); + assertThat(history).size().isOne(); + } + + class TestReactiveMessageHandler extends AbstractReactiveMessageHandler { + + @Override + protected Mono handleMessageInternal(Message message) { + + return Mono.fromSupplier(() -> { + handled.getAndSet(true); + output.send(message); + return handled.get(); + }).then(); + } + + } + +} 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 new file mode 100644 index 0000000000..0b067fed9b --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandler.java @@ -0,0 +1,131 @@ +/* + * Copyright 2019 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.outbound; + +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.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.convert.MongoConverter; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.handler.AbstractReactiveMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +import reactor.core.publisher.Mono; + +/** + * Implementation of {@link org.springframework.messaging.ReactiveMessageHandler} which writes + * Message payload into a MongoDb collection, using reactive MongoDb support, The + * collection is identified by evaluation of the {@link #collectionNameExpression}. + * + * @author David Turanski + * + * @since 5.3 + */ +public class ReactiveMongoDbStoringMessageHandler extends AbstractReactiveMessageHandler { + + private ReactiveMongoOperations mongoTemplate; + + private ReactiveMongoDatabaseFactory mongoDbFactory; + + private MongoConverter mongoConverter; + + private StandardEvaluationContext evaluationContext; + + private Expression collectionNameExpression = new LiteralExpression("data"); + + private volatile boolean initialized = false; + + /** + * Construct this instance using a provided {@link ReactiveMongoDatabaseFactory} + * @param mongoDbFactory The reactive mongoDatabase factory. + */ + public ReactiveMongoDbStoringMessageHandler(ReactiveMongoDatabaseFactory mongoDbFactory) { + Assert.notNull(mongoDbFactory, "'mongoDbFactory' must not be null"); + this.mongoDbFactory = mongoDbFactory; + } + + /** + * Construct this instance using a fully created and initialized instance of provided + * {@link ReactiveMongoOperations} + * @param mongoTemplate The ReactiveMongoOperations implementation. + */ + public ReactiveMongoDbStoringMessageHandler(ReactiveMongoOperations mongoTemplate) { + Assert.notNull(mongoTemplate, "'mongoTemplate' must not be null"); + this.mongoTemplate = mongoTemplate; + } + + /** + * Provide a custom {@link MongoConverter} used to assist in serialization of + * data written to MongoDb. Only allowed if this instance was constructed with a + * {@link MongoDbFactory}. + * @param mongoConverter The mongo converter. + */ + public void setMongoConverter(MongoConverter mongoConverter) { + Assert.isNull(this.mongoTemplate, + "'mongoConverter' can not be set when instance was constructed with MongoTemplate"); + this.mongoConverter = mongoConverter; + } + + /** + * Set a SpEL {@link Expression} that should resolve to a collection name used by + * {@link MongoOperations} to store data + * @param collectionNameExpression The collection name expression. + */ + public void setCollectionNameExpression(Expression collectionNameExpression) { + Assert.notNull(collectionNameExpression, "'collectionNameExpression' must not be null"); + this.collectionNameExpression = collectionNameExpression; + } + + @Override + public String getComponentType() { + return "mongo:reactive-outbound-channel-adapter"; + } + + @Override + protected void onInit() { + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + if (this.mongoTemplate == null) { + this.mongoTemplate = new ReactiveMongoTemplate(this.mongoDbFactory, this.mongoConverter); + } + this.initialized = true; + } + + @Override + protected Mono handleMessageInternal(Message message) { + Assert.isTrue(this.initialized, "This class is not yet initialized. Invoke its afterPropertiesSet() method"); + + return evaluateCollectionNameExpression(message) + .flatMap(collection -> this.mongoTemplate.save(message.getPayload(), collection)) + .then(); + } + + private Mono evaluateCollectionNameExpression(Message message) { + return Mono.fromSupplier(() -> { + String collectionName = + this.collectionNameExpression.getValue(this.evaluationContext, message, String.class); + Assert.notNull(collectionName, "'collectionNameExpression' must not evaluate to null"); + return collectionName; + }); + } + +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbStoringMessageHandlerTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbStoringMessageHandlerTests.java index 0c2fbcda72..d8288b696b 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbStoringMessageHandlerTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbStoringMessageHandlerTests.java @@ -17,12 +17,14 @@ 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.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import org.bson.conversions.Bson; +import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; @@ -49,29 +51,41 @@ import org.springframework.messaging.Message; */ public class MongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { - @Test(expected = IllegalArgumentException.class) - public void withNullMongoDBFactory() { - new MongoDbStoringMessageHandler((MongoDbFactory) null); - } + private MongoTemplate template; - @Test(expected = IllegalArgumentException.class) - public void withNullMongoTemplate() { - new MongoDbStoringMessageHandler((MongoOperations) null); + private MongoDbFactory mongoDbFactory; + + @Before + public void setUp() { + mongoDbFactory = prepareMongoFactory("foo"); + template = new MongoTemplate(mongoDbFactory); } @Test @MongoDbAvailable - public void validateMessageHandlingWithDefaultCollection() throws Exception { + public void withNullMongoDBFactory() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new MongoDbStoringMessageHandler((MongoDbFactory) null)); + } - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(mongoDbFactory); + @Test + @MongoDbAvailable + public void withNullMongoTemplate() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new MongoDbStoringMessageHandler((MongoOperations) null)); + } + + @Test + @MongoDbAvailable + public void validateMessageHandlingWithDefaultCollection() { + + MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(this.mongoDbFactory); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); handler.handleMessage(message); - MongoTemplate template = new MongoTemplate(mongoDbFactory); Query query = new BasicQuery("{'name' : 'Bob'}"); Person person = template.findOne(query, Person.class, "data"); @@ -81,17 +95,15 @@ public class MongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void validateMessageHandlingWithNamedCollection() throws Exception { + public void validateMessageHandlingWithNamedCollection() { - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(mongoDbFactory); + MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(this.mongoDbFactory); handler.setCollectionNameExpression(new LiteralExpression("foo")); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); handler.handleMessage(message); - MongoTemplate template = new MongoTemplate(mongoDbFactory); Query query = new BasicQuery("{'name' : 'Bob'}"); Person person = template.findOne(query, Person.class, "foo"); @@ -101,10 +113,9 @@ public class MongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void validateMessageHandlingWithMongoConverter() throws Exception { + public void validateMessageHandlingWithMongoConverter() { - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(mongoDbFactory); + MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(this.mongoDbFactory); handler.setCollectionNameExpression(new LiteralExpression("foo")); MappingMongoConverter converter = new TestMongoConverter(mongoDbFactory, new MongoMappingContext()); converter.afterPropertiesSet(); @@ -115,7 +126,6 @@ public class MongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); handler.handleMessage(message); - MongoTemplate template = new MongoTemplate(mongoDbFactory); Query query = new BasicQuery("{'name' : 'Bob'}"); Person person = template.findOne(query, Person.class, "foo"); @@ -126,27 +136,26 @@ public class MongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { @Test @MongoDbAvailable - public void validateMessageHandlingWithMongoTemplate() throws Exception { - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MappingMongoConverter converter = new TestMongoConverter(mongoDbFactory, new MongoMappingContext()); + public void validateMessageHandlingWithMongoTemplate() { + MappingMongoConverter converter = new TestMongoConverter(this.mongoDbFactory, new MongoMappingContext()); converter.afterPropertiesSet(); converter = spy(converter); - MongoTemplate template = new MongoTemplate(mongoDbFactory, converter); + MongoTemplate writingTemplate = new MongoTemplate(this.mongoDbFactory, converter); - MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(template); + MongoDbStoringMessageHandler handler = new MongoDbStoringMessageHandler(writingTemplate); handler.setCollectionNameExpression(new LiteralExpression("foo")); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); handler.handleMessage(message); - MongoTemplate readingTemplate = new MongoTemplate(mongoDbFactory); Query query = new BasicQuery("{'name' : 'Bob'}"); - Person person = readingTemplate.findOne(query, Person.class, "foo"); + Person person = template.findOne(query, Person.class, "foo"); assertThat(person.getName()).isEqualTo("Bob"); assertThat(person.getAddress().getState()).isEqualTo("PA"); verify(converter, times(1)).write(Mockito.any(), Mockito.any(Bson.class)); } + } 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 new file mode 100644 index 0000000000..29941931cd --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests.java @@ -0,0 +1,179 @@ +/* + * Copyright 2019 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.outbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.ReactiveMongoOperations; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.convert.MappingMongoConverter; +import org.springframework.data.mongodb.core.mapping.MongoMappingContext; +import org.springframework.data.mongodb.core.query.BasicQuery; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.expression.common.LiteralExpression; +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 reactor.core.publisher.Mono; + +/** + * @author Amol Nayak + * @author Oleg Zhurakousky + * @author Gary Russell + * @author David Turanski + * + * @since 5.3 + */ +public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableTests { + + private ReactiveMongoTemplate template; + + private ReactiveMongoDatabaseFactory mongoDbFactory; + + @Before + public void setUp() { + mongoDbFactory = this.prepareReactiveMongoFactory("foo"); + template = new ReactiveMongoTemplate(mongoDbFactory); + } + + @Test + @MongoDbAvailable + public void withNullMongoDBFactory() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new ReactiveMongoDbStoringMessageHandler((ReactiveMongoDatabaseFactory) null)); + } + + @Test + @MongoDbAvailable + public void withNullMongoTemplate() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new ReactiveMongoDbStoringMessageHandler((ReactiveMongoOperations) null)); + } + + @Test + @MongoDbAvailable + public void validateMessageHandlingWithDefaultCollection() { + ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); + waitFor(handler.handleMessage(message)); + + Query query = new BasicQuery("{'name' : 'Bob'}"); + Person person = waitFor(this.template.findOne(query, Person.class, "data")); + assertThat(person.getName()).isEqualTo("Bob"); + assertThat(person.getAddress().getState()).isEqualTo("PA"); + + } + + @Test + @MongoDbAvailable + public void validateMessageHandlingWithNamedCollection() { + ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); + handler.setCollectionNameExpression(new LiteralExpression("foo")); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); + waitFor(handler.handleMessage(message)); + + Query query = new BasicQuery("{'name' : 'Bob'}"); + Person person = waitFor(this.template.findOne(query, Person.class, "foo")); + + assertThat(person.getName()).isEqualTo("Bob"); + assertThat(person.getAddress().getState()).isEqualTo("PA"); + + } + + @Test + @MongoDbAvailable + public void errorOnMessageHandlingWithNullValuedExpression() { + + ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); + handler.setCollectionNameExpression(new LiteralExpression(null)); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload(createPerson("Bob")).build(); + AtomicBoolean errorOccurred = new AtomicBoolean(); + handler.handleMessage(message) + .doOnError(e -> errorOccurred.set(true)) + .subscribe(aVoid -> assertThat(errorOccurred.get()).isTrue()); + } + + @Test + @MongoDbAvailable + public void validateMessageHandlingWithMongoConverter() { + ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory); + handler.setCollectionNameExpression(new LiteralExpression("foo")); + MappingMongoConverter converter = + new ReactiveTestMongoConverter(this.mongoDbFactory, new MongoMappingContext()); + converter.afterPropertiesSet(); + converter = spy(converter); + handler.setMongoConverter(converter); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); + waitFor(handler.handleMessage(message)); + + Query query = new BasicQuery("{'name' : 'Bob'}"); + Person person = waitFor(this.template.findOne(query, Person.class, "foo")); + + assertThat(person.getName()).isEqualTo("Bob"); + assertThat(person.getAddress().getState()).isEqualTo("PA"); + } + + @Test + @MongoDbAvailable + public void validateMessageHandlingWithMongoTemplate() { + MappingMongoConverter converter = + new ReactiveTestMongoConverter(this.mongoDbFactory, new MongoMappingContext()); + converter.afterPropertiesSet(); + converter = spy(converter); + ReactiveMongoTemplate writingTemplate = new ReactiveMongoTemplate(this.mongoDbFactory, converter); + ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(writingTemplate); + handler.setCollectionNameExpression(new LiteralExpression("foo")); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + Message message = MessageBuilder.withPayload(this.createPerson("Bob")).build(); + waitFor(handler.handleMessage(message)); + + Query query = new BasicQuery("{'name' : 'Bob'}"); + Person person = waitFor(this.template.findOne(query, Person.class, "foo")); + + assertThat(person.getName()).isEqualTo("Bob"); + assertThat(person.getAddress().getState()).isEqualTo("PA"); + } + + private static T waitFor(Mono mono) { + return mono.block(Duration.ofSeconds(3)); + } + +} 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 f0978f8836..8df8a9c232 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 @@ -16,6 +16,8 @@ package org.springframework.integration.mongodb.rules; +import java.time.Duration; + import org.bson.Document; import org.bson.conversions.Bson; import org.junit.Rule; @@ -23,10 +25,14 @@ import org.junit.Rule; import org.springframework.dao.DataAccessException; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.data.mongodb.core.SimpleMongoClientDbFactory; +import org.springframework.data.mongodb.core.SimpleReactiveMongoDatabaseFactory; import org.springframework.data.mongodb.core.convert.DefaultDbRefResolver; import org.springframework.data.mongodb.core.convert.MappingMongoConverter; +import org.springframework.data.mongodb.core.convert.NoOpDbRefResolver; import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity; import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty; import org.springframework.integration.mongodb.outbound.MessageCollectionCallback; @@ -36,19 +42,21 @@ import com.mongodb.MongoException; import com.mongodb.client.MongoClients; import com.mongodb.client.MongoCollection; + /** * Convenience base class that enables unit test methods to rely upon the {@link MongoDbAvailable} annotation. * * @author Oleg Zhurakousky * @author Xavier Padro * @author Artem Bilan + * @author David Turanski * * @since 2.1 */ public abstract class MongoDbAvailableTests { @Rule - public MongoDbAvailableRule redisAvailableRule = new MongoDbAvailableRule(); + public MongoDbAvailableRule mongoDbAvailableRule = new MongoDbAvailableRule(); protected MongoDbFactory prepareMongoFactory(String... additionalCollectionsToDrop) { @@ -57,6 +65,25 @@ public abstract class MongoDbAvailableTests { return mongoDbFactory; } + protected ReactiveMongoDatabaseFactory prepareReactiveMongoFactory(String... additionalCollectionsToDrop) { + ReactiveMongoDatabaseFactory mongoDbFactory = new SimpleReactiveMongoDatabaseFactory( + com.mongodb.reactivestreams.client.MongoClients.create(), "test"); + cleanupCollections(mongoDbFactory, additionalCollectionsToDrop); + return mongoDbFactory; + } + + protected void cleanupCollections(ReactiveMongoDatabaseFactory mongoDbFactory, + String... additionalCollectionsToDrop) { + + ReactiveMongoTemplate template = new ReactiveMongoTemplate(mongoDbFactory); + template.dropCollection("messages").block(Duration.ofSeconds(3)); + template.dropCollection("configurableStoreMessages").block(Duration.ofSeconds(3)); + template.dropCollection("data").block(Duration.ofSeconds(3)); + for (String additionalCollection : additionalCollectionsToDrop) { + template.dropCollection(additionalCollection).block(Duration.ofSeconds(3)); + } + } + protected void cleanupCollections(MongoDbFactory mongoDbFactory, String... additionalCollectionsToDrop) { MongoTemplate template = new MongoTemplate(mongoDbFactory); template.dropCollection("messages"); @@ -170,6 +197,27 @@ public abstract class MongoDbAvailableTests { } + public static class ReactiveTestMongoConverter extends MappingMongoConverter { + + public ReactiveTestMongoConverter( + ReactiveMongoDatabaseFactory mongoDbFactory, + MappingContext, MongoPersistentProperty> mappingContext) { + + super(NoOpDbRefResolver.INSTANCE, mappingContext); + } + + @Override + public void write(Object source, Bson target) { + super.write(source, target); + } + + @Override + public S read(Class clazz, Bson source) { + return super.read(clazz, source); + } + + } + public static class TestCollectionCallback implements MessageCollectionCallback { @Override diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 4f169d9765..1a781333b0 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -23,3 +23,5 @@ See its JavaDocs and <<./graph.adoc#integration-graph,Integration Graph>> for mo [[x5.3-general]] === General Changes + +