Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix IntegrationFlowDefinition.toReactivePublisher

Browse Source 
https://build.spring.io/browse/INT-MASTER-478

The `EmitterProcessor` proves to be unstable in between subscribers which come and go away from time to time.
Looks like buffered events are dropped when we don't have subscriber just after the previous one has been unsubscribed

* Change the logic in the `IntegrationFlowDefinition.toReactivePublisher()` to call `ReactiveConsumer.adaptToPublisher(MessageChannel)` directly to return provided `Publisher<?>` to the caller for his own responsibility.
This way we don't have any buffering or prefetching in the Framework because we don't do any `Subscription.request(n)` explicitly
* Prove the fix with polishing to the `ReactiveStreamsTests.testPollableReactiveFlow()`
Right now we don't poll `QueueChannel` if there is no demand, therefore we don't need extra `CountDownLatch` to wait for second subscriber.
Plus the logic now remains as a `queue` manner: consumers are concurrent for the data in the `QueueChannel`
Extra `@Repeat(10)` for confirmation.
Previously it failed sporadically, what is confirmed with that one more CI failure
This commit is contained in:
Artem Bilan
2016-12-20 17:39:24 -05:00
parent 37b9612f24
commit 9a6aa0dbef
3 changed files with 9 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -25,9 +25,8 @@ import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.Function;
import org.reactivestreams.Processor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.BeanCreationException;
@@ -96,7 +95,6 @@ import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.publisher.EmitterProcessor;
import reactor.util.function.Tuple2;
/**
@@ -2728,10 +2726,8 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
else {
if (channelForPublisher != null) {
Processor<?, ?> processor = EmitterProcessor.create(false);
publisher = (Publisher<Message<T>>) processor;
Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
addComponent(new ReactiveConsumer(channelForPublisher, subscriber));
Publisher<?> messagePublisher = ReactiveConsumer.adaptToPublisher(channelForPublisher);
publisher = (Publisher<Message<T>>) messagePublisher;
}
else {
MessageChannel reactiveChannel = new ReactiveChannel();
@@ -2742,7 +2738,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
get();
return new PublisherIntegrationFlow<T>(this.integrationComponents, publisher);
return new PublisherIntegrationFlow<>(this.integrationComponents, publisher);
}
private <S extends ConsumerEndpointSpec<S, ? extends MessageHandler>> B register(S endpointSpec,

6
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java
View File

@@ -111,7 +111,7 @@ public class ReactiveConsumer extends AbstractEndpoint {
this.subscriber.cancel();
}
private Publisher<Message<?>> adaptToPublisher(MessageChannel inputChannel) {
public static Publisher<Message<?>> adaptToPublisher(MessageChannel inputChannel) {
if (inputChannel instanceof SubscribableChannel) {
return adaptSubscribableChannelToPublisher((SubscribableChannel) inputChannel);
}
@@ -124,11 +124,11 @@ public class ReactiveConsumer extends AbstractEndpoint {
}
}
private Publisher<Message<?>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
private static Publisher<Message<?>> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) {
return new SubscribableChannelPublisherAdapter(inputChannel);
}
private Publisher<Message<?>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
private static Publisher<Message<?>> adaptPollableChannelToPublisher(PollableChannel inputChannel) {
return new PollableChannelPublisherAdapter(inputChannel);
}

7
spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
View File

@@ -52,6 +52,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.Repeat;
import org.springframework.test.context.junit4.SpringRunner;
import reactor.core.publisher.Flux;
@@ -103,6 +104,7 @@ public class ReactiveStreamsTests {
}
@Test
@Repeat(10)
public void testPollableReactiveFlow() throws Exception {
this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
@@ -115,8 +117,6 @@ public class ReactiveStreamsTests {
.doOnNext(p -> latch.countDown())
.subscribe();
CountDownLatch secondSubscriberLatch = new CountDownLatch(1);
Future<List<Integer>> future =
Executors.newSingleThreadExecutor().submit(() ->
Flux.just("11,12,13")
@@ -129,12 +129,9 @@ public class ReactiveStreamsTests {
.map(Message::getPayload)
.log("org.springframework.integration.flux")
.collectList()
.doOnRequest(s -> secondSubscriberLatch.countDown())
.block(Duration.ofSeconds(10))
);
assertTrue(secondSubscriberLatch.await(10, TimeUnit.SECONDS));
this.inputChannel.send(new GenericMessage<>("6,7,8,9,10"));
assertTrue(latch.await(10, TimeUnit.SECONDS));