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

Honor the FluxSink State in the PollChPublisherAd

Browse Source 
The `PollableChannelPublisherAdapter` is based on the poll model of the
`FluxSink` and iterate and poll downstream `PollableChannel` until
there is an item or `n > 0`.
Having `take(6)` we end up with the cancel from the downstream
`Subscriber` after the batch is filled, but at the same time we continue
to poll the upstream source because `n` is like `Long.MAX_VALUE`.

* The proper way to interact is check for the `!sink.isCancelled()`
as well.
This way cancelled `sink` won't "steal" data from other subscribers

Fix compatibility with the latest dependencies

* Upgrade to Gradle 4.1
* Upgrade as much dependencies as possible
* Fix MongoDB module to resolve deprecation in the latest driver
* Increase receive timeout in the `ResequencerTests`
* Restore generic argument for the method reference in the `ReactiveStreamsTests`
* Fix `WebFluxInboundEndpoint` for the compatibility with the latest Reactor API
This commit is contained in:
Artem Bilan
2017-09-01 16:26:50 -04:00
committed by Gary Russell
parent cfab1fb1a4
commit 46de69dbf2
10 changed files with 74 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannelReactiveUtils.java
View File

@@ -106,7 +106,7 @@ public final class MessageChannelReactiveUtils {
.<Message<T>>create(sink ->
sink.onRequest(n -> {
Message<?> m;
while (n-- > 0 && (m = this.channel.receive()) != null) {
while (!sink.isCancelled() && n-- > 0 && (m = this.channel.receive()) != null) {
sink.next((Message<T>) m);
}
}),

7
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -262,8 +262,9 @@ public class ResequencerTests {
assertNotNull(reply1);
assertNotNull(reply2);
assertNull(reply3);
ArrayList<Integer> sequence = new ArrayList<Integer>(Arrays.asList(new IntegrationMessageHeaderAccessor(reply1).getSequenceNumber(),
new IntegrationMessageHeaderAccessor(reply2).getSequenceNumber()));
ArrayList<Integer> sequence = new ArrayList<>(
Arrays.asList(new IntegrationMessageHeaderAccessor(reply1).getSequenceNumber(),
new IntegrationMessageHeaderAccessor(reply2).getSequenceNumber()));
Collections.sort(sequence);
assertEquals("[1, 2]", sequence.toString());
// Once a group is expired, late messages are discarded immediately by default
@@ -363,7 +364,7 @@ public class ResequencerTests {
this.resequencer.handleMessage(message2);
Message<?> out1 = replyChannel.receive(10);
assertNull(out1);
out1 = discardChannel.receive(1000);
out1 = discardChannel.receive(10000);
assertNotNull(out1);
Message<?> out2 = discardChannel.receive(10);
assertNotNull(out2);

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

@@ -33,7 +33,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.reactivestreams.Publisher;
@@ -105,7 +104,6 @@ public class ReactiveStreamsTests {
}
@Test
@Ignore
public void testPollableReactiveFlow() throws Exception {
this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));