From effca79f3f9c83070cbc49cffd8d24997a8422d0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 14 Apr 2017 17:30:05 -0400 Subject: [PATCH] Upgrade to Reactor 3.1 Polishing - remove catch from test - update netty for stomp Add reactiveErrorHandler to AbstractMessageHandler --- build.gradle | 4 +-- .../integration/channel/ReactiveChannel.java | 16 ++++++---- .../integration/dsl/Channels.java | 8 ++--- .../dsl/channel/MessageChannels.java | 9 +++--- .../dsl/channel/ReactiveChannelSpec.java | 7 ++-- .../handler/AbstractMessageHandler.java | 29 ++++++++++++++++- .../reactive/ReactiveChannelTests.java | 32 +++++++++++-------- 7 files changed, 70 insertions(+), 35 deletions(-) diff --git a/build.gradle b/build.gradle index 1ab8a1f4c7..2c9951ccde 100644 --- a/build.gradle +++ b/build.gradle @@ -124,8 +124,8 @@ subprojects { subproject -> mysqlVersion = '5.1.34' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' - reactorNettyVersion = '0.6.2.RELEASE' - reactorVersion = '3.0.6.RELEASE' + reactorNettyVersion = '0.6.3.BUILD-SNAPSHOT' + reactorVersion = '3.1.0.BUILD-SNAPSHOT' romeToolsVersion = '1.7.0' servletApiVersion = '3.1.0' slf4jVersion = "1.7.21" diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java index 52e19bce74..1d3d902891 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java @@ -20,19 +20,20 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; -import org.reactivestreams.Processor; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.springframework.messaging.Message; import org.springframework.util.Assert; -import reactor.core.publisher.BlockingSink; import reactor.core.publisher.DirectProcessor; import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxProcessor; +import reactor.core.publisher.FluxSink; /** * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -43,11 +44,11 @@ public class ReactiveChannel extends AbstractMessageChannel private final List>> publishers = new CopyOnWriteArrayList<>(); - private final Processor, Message> processor; + private final FluxProcessor, Message> processor; private final Flux> flux; - private final BlockingSink> sink; + private final FluxSink> sink; private volatile boolean upstreamSubscribed; @@ -55,16 +56,17 @@ public class ReactiveChannel extends AbstractMessageChannel this(DirectProcessor.create()); } - public ReactiveChannel(Processor, Message> processor) { + public ReactiveChannel(FluxProcessor, Message> processor) { Assert.notNull(processor, "'processor' must not be null"); this.processor = processor; this.flux = Flux.from(processor); - this.sink = BlockingSink.create(this.processor); + this.sink = processor.sink(); } @Override protected boolean doSend(Message message, long timeout) { - return this.sink.submit(message, timeout) > -1; + this.sink.next(message); + return true; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java index 8a21be4fb9..a67d2af349 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java @@ -19,8 +19,6 @@ package org.springframework.integration.dsl; import java.util.Queue; import java.util.concurrent.Executor; -import org.reactivestreams.Processor; - import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.ExecutorChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; @@ -33,6 +31,8 @@ import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.messaging.Message; +import reactor.core.publisher.FluxProcessor; + /** * @author Artem Bilan * @author Gary Russell @@ -140,11 +140,11 @@ public class Channels { return MessageChannels.reactive(id); } - public ReactiveChannelSpec reactive(Processor, Message> processor) { + public ReactiveChannelSpec reactive(FluxProcessor, Message> processor) { return MessageChannels.reactive(processor); } - public ReactiveChannelSpec reactive(String id, Processor, Message> processor) { + public ReactiveChannelSpec reactive(String id, FluxProcessor, Message> processor) { return MessageChannels.reactive(id, processor); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index 9fd2c590e0..48391dea4d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -19,14 +19,15 @@ package org.springframework.integration.dsl.channel; import java.util.Queue; import java.util.concurrent.Executor; -import org.reactivestreams.Processor; - import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; import org.springframework.messaging.Message; +import reactor.core.publisher.FluxProcessor; + /** * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -133,11 +134,11 @@ public final class MessageChannels { return reactive().id(id); } - public static ReactiveChannelSpec reactive(Processor, Message> processor) { + public static ReactiveChannelSpec reactive(FluxProcessor, Message> processor) { return new ReactiveChannelSpec(processor); } - public static ReactiveChannelSpec reactive(String id, Processor, Message> processor) { + public static ReactiveChannelSpec reactive(String id, FluxProcessor, Message> processor) { return reactive(processor).id(id); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java index ab85627199..c347d6d04b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/ReactiveChannelSpec.java @@ -16,13 +16,14 @@ package org.springframework.integration.dsl.channel; -import org.reactivestreams.Processor; - import org.springframework.integration.channel.ReactiveChannel; import org.springframework.messaging.Message; +import reactor.core.publisher.FluxProcessor; + /** * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -32,7 +33,7 @@ public class ReactiveChannelSpec extends MessageChannelSpec, Message> processor) { + ReactiveChannelSpec(FluxProcessor, Message> processor) { this.channel = new ReactiveChannel(processor); } 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 3144b06285..aa4c9d6d42 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 @@ -20,6 +20,7 @@ import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import org.springframework.core.Ordered; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.Orderable; import org.springframework.integration.history.MessageHistory; @@ -36,6 +37,7 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; +import org.springframework.util.ErrorHandler; import reactor.core.publisher.Operators; @@ -70,6 +72,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im private volatile boolean loggingEnabled = true; + private ErrorHandler reactiveErrorHandler; + @Override public boolean isLoggingEnabled() { return this.loggingEnabled; @@ -100,6 +104,16 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im this.shouldTrack = shouldTrack; } + /** + * Set the error handler to use when an exception occurs when this handler + * is invoked as a reactive {@link Subscriber}. + * @param reactiveErrorHandler the error handler. + * @since 5.0 + */ + public void setReactiveErrorHandler(ErrorHandler reactiveErrorHandler) { + this.reactiveErrorHandler = reactiveErrorHandler; + } + @Override public void configureMetrics(AbstractMessageHandlerMetrics metrics) { Assert.notNull(metrics, "'metrics' must not be null"); @@ -111,6 +125,14 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im if (this.statsEnabled) { this.handlerMetrics.setFullStatsEnabled(true); } + if (this.reactiveErrorHandler == null) { + if (getBeanFactory() != null) { + this.reactiveErrorHandler = new MessagePublishingErrorHandler(getChannelResolver()); + } + else { + this.reactiveErrorHandler = new MessagePublishingErrorHandler(); + } + } } @Override @@ -154,7 +176,12 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im @Override public void onNext(Message message) { - handleMessage(message); + try { + handleMessage(message); + } + catch (MessagingException e) { + this.reactiveErrorHandler.handleError(e); + } } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java index 5baa1ff17e..eac9b4f7d3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveChannelTests.java @@ -17,10 +17,10 @@ package org.springframework.integration.channel.reactive; import static org.hamcrest.Matchers.contains; -import static org.hamcrest.Matchers.containsString; -import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.isOneOf; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -42,8 +42,8 @@ import org.springframework.integration.channel.ReactiveChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageDeliveryException; -import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.annotation.DirtiesContext; @@ -65,21 +65,15 @@ public class ReactiveChannelTests { @Autowired private MessageChannel queueChannel; + @Autowired + private PollableChannel errorChannel; + @Test - @SuppressWarnings("unchecked") public void testReactiveMessageChannel() throws InterruptedException { QueueChannel replyChannel = new QueueChannel(); for (int i = 0; i < 10; i++) { - try { - this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); - } - catch (Exception e) { - assertThat(e, instanceOf(MessageDeliveryException.class)); - assertThat(e.getCause().getCause(), instanceOf(MessageHandlingException.class)); - assertThat(e.getCause().getCause().getCause(), instanceOf(IllegalStateException.class)); - assertThat(e.getMessage(), containsString("intentional")); - } + this.reactiveChannel.send(MessageBuilder.withPayload(i).setReplyChannel(replyChannel).build()); } for (int i = 0; i < 9; i++) { @@ -87,6 +81,11 @@ public class ReactiveChannelTests { assertNotNull(receive); assertThat(receive.getPayload(), isOneOf("0", "1", "2", "3", "4", "6", "7", "8", "9")); } + assertNull(replyChannel.receive(0)); + + Message error = this.errorChannel.receive(0); + assertNotNull(error); + assertEquals(5, ((MessagingException) error.getPayload()).getFailedMessage().getPayload()); } @Test @@ -111,6 +110,11 @@ public class ReactiveChannelTests { @EnableIntegration public static class TestConfiguration { + @Bean + public QueueChannel errorChannel() { + return new QueueChannel(); + } + @Bean public MessageChannel reactiveChannel() { return new ReactiveChannel();