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

Use EmitterProcessor do not drop polled messages

Browse Source 
The `BridgeHandler` is Reactive `Subscriber` with `MAX_VALUE` demand just `onSubscribe()`.
That causes a drain of the upstream `QueueChannel` to the `ReactiveChannel` with `DirectProcessor`.
 The last one just drops messages if there is no subscribers

`EmitterProcessor` doesn't request upstream until real subscriber arrives to it.
Therefore change `BridgeHandler` logic in the `IntegrationFlowDefinition.toReactivePublisher()` to the `EmitterProcessor` to allow late `Subscriber`s and don't lose messages from the `QueueChannel` if there is no downstream `Subscriber`s
This commit is contained in:
Artem Bilan
2016-11-29 11:58:59 -05:00
parent f9789b9e67
commit 7be7968e05
2 changed files with 15 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,8 +25,9 @@ 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;
@@ -95,6 +96,7 @@ 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;
/**
@@ -2725,21 +2727,17 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
publisher = (Publisher<Message<T>>) channelForPublisher;
}
else {
MessageChannel reactiveChannel = new ReactiveChannel();
publisher = (Publisher<Message<T>>) reactiveChannel;
if (channelForPublisher != null) {
BridgeHandler bridge = new BridgeHandler();
bridge.setOutputChannel(reactiveChannel);
addComponent(bridge)
.addComponent(new ReactiveConsumer(channelForPublisher, bridge))
.addComponent(reactiveChannel);
Processor<?, ?> processor = EmitterProcessor.create(false);
publisher = (Publisher<Message<T>>) processor;
Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
addComponent(new ReactiveConsumer(channelForPublisher, subscriber));
}
else {
MessageChannel reactiveChannel = new ReactiveChannel();
publisher = (Publisher<Message<T>>) reactiveChannel;
channel(reactiveChannel);
}
}
get();

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

@@ -39,7 +39,6 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.Lifecycle;
@@ -107,16 +106,16 @@ public class ReactiveStreamsTests {
public void testPollableReactiveFlow() throws InterruptedException, TimeoutException, ExecutionException {
this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
CountDownLatch warmUpLatch = new CountDownLatch(3);
CountDownLatch latch = new CountDownLatch(6);
Flux.from(this.pollablePublisher)
.filter(m -> m.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
.doOnNext(p -> latch.countDown())
.doOnNext(p -> warmUpLatch.countDown())
.take(6)
.subscribe();
final CountDownLatch asyncSetupLatch = new CountDownLatch(1);
Future<List<Integer>> future =
Executors.newSingleThreadExecutor().submit(() ->
Flux.just("11,12,13")
@@ -127,23 +126,21 @@ public class ReactiveStreamsTests {
.concatWith(this.pollablePublisher)
.map(Message::getPayload)
.take(7)
.log("org.springframework.integration.flux")
.collectList()
.block(getDuration(asyncSetupLatch)));
.block(Duration.ofSeconds(10)));
assertTrue(asyncSetupLatch.await(10, TimeUnit.SECONDS));
assertTrue(warmUpLatch.await(10, TimeUnit.SECONDS));
this.inputChannel.send(new GenericMessage<>("6,7,8,9,10"));
assertTrue(latch.await(10, TimeUnit.SECONDS));
List<Integer> integers = future.get(20, TimeUnit.SECONDS);
assertNotNull(integers);
assertEquals(7, integers.size());
}
private Duration getDuration(CountDownLatch asyncSetupLatch) {
asyncSetupLatch.countDown();
return Duration.ofSeconds(10);
}
@Configuration
@EnableIntegration