From b79e4d93a56418e6edb1b6bedbb4313f5e84cd41 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 12 Dec 2016 18:18:45 -0500 Subject: [PATCH] Use TopicProcessor for PublisherIntegrationFlow https://build.spring.io/browse/INT-MASTER-467/ Even if `EmitterProcessor` looks good for us not requesting until real `subscriber()`, it doesn't keep backlog for late subscribers like it happens some times in our subscriber in the separate Thread. The solution should be considered as tentative because the real one must be based on the fact that `AMPH` has to perform `subscription.request(n)` as a fact of the subscription from downstream. In other words mid-flow components should work as passive subscribers and be based on downstream demand --- .../integration/dsl/IntegrationFlowDefinition.java | 4 ++-- .../integration/config/ChannelAdapterParserTests.java | 8 +++++++- .../dsl/reactivestreams/ReactiveStreamsTests.java | 11 +++++------ 3 files changed, 14 insertions(+), 9 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 7a4c9600c6..0f6d9aa7ab 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -96,7 +96,7 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; -import reactor.core.publisher.EmitterProcessor; +import reactor.core.publisher.TopicProcessor; import reactor.util.function.Tuple2; /** @@ -2728,7 +2728,7 @@ public abstract class IntegrationFlowDefinition processor = EmitterProcessor.create(false); + Processor processor = TopicProcessor.share(false); publisher = (Publisher>) processor; Subscriber> subscriber = (Subscriber>) processor; addComponent(new ReactiveConsumer(channelForPublisher, subscriber)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java index d7d89c2155..c12709d842 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java @@ -55,6 +55,7 @@ import org.springframework.messaging.support.GenericMessage; public class ChannelAdapterParserTests { private AbstractApplicationContext applicationContext; + private AbstractApplicationContext applicationContextInner; @@ -69,6 +70,7 @@ public class ChannelAdapterParserTests { @After public void tearDown() { this.applicationContext.close(); + this.applicationContextInner.close(); } @@ -307,17 +309,21 @@ public class ChannelAdapterParserTests { assertEquals("test", message.getPayload()); MessageSource testMessageSource = this.applicationContext.getBean("testMessageSource", MessageSource.class); - SourcePollingChannelAdapter adapterWithMessageSourceRef = this.applicationContext.getBean("adapterWithMessageSourceRef", SourcePollingChannelAdapter.class); + SourcePollingChannelAdapter adapterWithMessageSourceRef = + this.applicationContext.getBean("adapterWithMessageSourceRef", SourcePollingChannelAdapter.class); MessageSource source = TestUtils.getPropertyValue(adapterWithMessageSourceRef, "source", MessageSource.class); assertSame(testMessageSource, source); } public static class SampleBean { + private final String message = "hello"; String getMessage() { return message; } + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java index 8f256c92c5..76a3cf443a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java @@ -27,11 +27,9 @@ import java.util.Arrays; import java.util.Date; import java.util.List; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.log4j.Level; @@ -79,7 +77,7 @@ public class ReactiveStreamsTests { private Publisher> pollablePublisher; @Autowired - @Qualifier("reactiveSteamsMessageSource") + @Qualifier("reactiveStreamsMessageSource") private Lifecycle messageSource; @Autowired @@ -87,7 +85,7 @@ public class ReactiveStreamsTests { private MessageChannel inputChannel; @Test - public void testReactiveFlow() throws InterruptedException { + public void testReactiveFlow() throws Exception { List results = new ArrayList<>(); CountDownLatch latch = new CountDownLatch(6); Flux.from(this.publisher) @@ -100,10 +98,11 @@ public class ReactiveStreamsTests { assertTrue(latch.await(10, TimeUnit.SECONDS)); String[] strings = results.toArray(new String[results.size()]); assertArrayEquals(new String[] { "A", "B", "C", "D", "E", "F" }, strings); + this.messageSource.stop(); } @Test - public void testPollableReactiveFlow() throws InterruptedException, TimeoutException, ExecutionException { + public void testPollableReactiveFlow() throws Exception { this.inputChannel.send(new GenericMessage<>("1,2,3,4,5")); CountDownLatch warmUpLatch = new CountDownLatch(3); @@ -154,7 +153,7 @@ public class ReactiveStreamsTests { .from(() -> new GenericMessage<>("a,b,c,d,e,f"), e -> e.poller(p -> p.trigger(ctx -> this.invoked.getAndSet(true) ? null : new Date())) .autoStartup(false) - .id("reactiveSteamsMessageSource")) + .id("reactiveStreamsMessageSource")) .split(String.class, p -> p.split(",")) .toReactivePublisher(); }