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

Revert to EmitterProcessor

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

Looks like demand in the `FluxTake` isn't honored:

```
public void request(long n) {
if (wip != 0) {
			s.request(n);
		}
		else if (WIP.compareAndSet(this, 0, 1)) {
				if (n >= this.n) {
					s.request(Long.MAX_VALUE);
				}
				else {
					s.request(n);
				}
		}
}
````
It isn't clear why we request from upstream `Long.MAX_VALUE`, if we have only strong `n` limit

So, revert to the `secondSubscriberLatch` and don't send the next message until the subscription from the second `Flux`
This commit is contained in:
Artem Bilan
2016-12-12 19:54:32 -05:00
parent b79e4d93a5
commit 61e77435ab
2 changed files with 23 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -96,7 +96,7 @@ import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.publisher.TopicProcessor;
import reactor.core.publisher.EmitterProcessor;
import reactor.util.function.Tuple2;
/**
@@ -2728,7 +2728,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
else {
if (channelForPublisher != null) {
Processor<?, ?> processor = TopicProcessor.share(false);
Processor<?, ?> processor = EmitterProcessor.create(false);
publisher = (Publisher<Message<T>>) processor;
Subscriber<Message<?>> subscriber = (Subscriber<Message<?>>) processor;
addComponent(new ReactiveConsumer(channelForPublisher, subscriber));

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

@@ -105,31 +105,35 @@ public class ReactiveStreamsTests {
public void testPollableReactiveFlow() throws Exception {
this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
CountDownLatch warmUpLatch = new CountDownLatch(3);
CountDownLatch latch = new CountDownLatch(6);
Flux.from(this.pollablePublisher)
.take(6)
.filter(m -> m.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
.doOnNext(p -> latch.countDown())
.doOnNext(p -> warmUpLatch.countDown())
.take(6)
.subscribe();
Future<List<Integer>> future =
Executors.newSingleThreadExecutor().submit(() ->
Flux.just("11,12,13")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
.map(Integer::parseInt)
.<Message<Integer>>map(GenericMessage<Integer>::new)
.concatWith(this.pollablePublisher)
.map(Message::getPayload)
.take(7)
.log("org.springframework.integration.flux")
.collectList()
.block(Duration.ofSeconds(10)));
CountDownLatch secondSubscriberLatch = new CountDownLatch(1);
assertTrue(warmUpLatch.await(10, TimeUnit.SECONDS));
Future<List<Integer>> future =
Executors.newSingleThreadExecutor().submit(() -> {
Thread.sleep(100);
return Flux.just("11,12,13")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
.map(Integer::parseInt)
.<Message<Integer>>map(GenericMessage<Integer>::new)
.concatWith(this.pollablePublisher)
.take(7)
.map(Message::getPayload)
.log("org.springframework.integration.flux")
.collectList()
.doOnSubscribe(s -> secondSubscriberLatch.countDown())
.block(Duration.ofSeconds(10));
}
);
assertTrue(secondSubscriberLatch.await(10, TimeUnit.SECONDS));
this.inputChannel.send(new GenericMessage<>("6,7,8,9,10"));