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

Fix pub/sub race conditions in Reactive tests

Browse Source
This commit is contained in:
Artem Bilan
2019-12-02 14:03:12 -05:00
parent daa89bf091
commit 8e7c1067f3
3 changed files with 71 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -152,11 +152,12 @@ public class ReactiveStreamsTests {
@Test
void testFromPublisher() {
Flux<Message<?>> messageFlux = Flux.just("1,2,3,4")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
.map(Integer::parseInt)
.map(GenericMessage::new);
Flux<Message<?>> messageFlux =
Flux.just("1,2,3,4")
.map(v -> v.split(","))
.flatMapIterable(Arrays::asList)
.map(Integer::parseInt)
.map(GenericMessage::new);
QueueChannel resultChannel = new QueueChannel();

69
spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java
View File

@@ -161,63 +161,74 @@ class DefaultSplitterTests {
void splitArrayPayloadReactive() {
Message<?> message = new GenericMessage<>(new String[] { "x", "y", "z" });
FluxMessageChannel replyChannel = new FluxMessageChannel();
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(replyChannel);
splitter.handleMessage(message);
Flux<String> testFlux =
Flux.from(replyChannel)
.map(Message::getPayload)
.cast(String.class);
StepVerifier.create(testFlux)
.expectNext("x", "y", "z")
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verify(Duration.ofSeconds(1));
StepVerifier verifier =
StepVerifier.create(testFlux)
.expectNext("x", "y", "z")
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verifyLater();
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(replyChannel);
splitter.handleMessage(message);
verifier.verify(Duration.ofSeconds(1));
}
@Test
void splitStreamReactive() {
Message<?> message = new GenericMessage<>(Stream.of("x", "y", "z"));
FluxMessageChannel replyChannel = new FluxMessageChannel();
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(replyChannel);
splitter.handleMessage(message);
Flux<String> testFlux =
Flux.from(replyChannel)
.map(Message::getPayload)
.cast(String.class);
StepVerifier.create(testFlux)
.expectNext("x", "y", "z")
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verify(Duration.ofSeconds(1));
StepVerifier verifier =
StepVerifier.create(testFlux)
.expectNext("x", "y", "z")
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verifyLater();
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(replyChannel);
splitter.handleMessage(message);
verifier.verify(Duration.ofSeconds(1));
}
@Test
void splitFluxReactive() {
Message<?> message = new GenericMessage<>(Flux.just("x", "y", "z"));
FluxMessageChannel replyChannel = new FluxMessageChannel();
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(replyChannel);
splitter.handleMessage(message);
Flux<String> testFlux =
Flux.from(replyChannel)
.map(Message::getPayload)
.cast(String.class);
StepVerifier.create(testFlux)
.expectNext("x", "y", "z")
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verify(Duration.ofSeconds(1));
StepVerifier verifier =
StepVerifier.create(testFlux)
.expectNext("x", "y", "z")
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verifyLater();
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(replyChannel);
splitter.handleMessage(message);
verifier.verify(Duration.ofSeconds(1));
}
}

52
spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java
View File

@@ -58,9 +58,7 @@ import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.context.support.AnnotationConfigContextLoader;
import org.springframework.util.FileCopyUtils;
import reactor.test.StepVerifier;
@@ -72,8 +70,7 @@ import reactor.test.StepVerifier;
*
* @since 4.1.2
*/
@ContextConfiguration(loader = AnnotationConfigContextLoader.class)
@SpringJUnitConfig
@SpringJUnitConfig(FileSplitterTests.ContextConfiguration.class)
@DirtiesContext
public class FileSplitterTests {
@@ -265,34 +262,35 @@ public class FileSplitterTests {
@Test
void testFileSplitterReactive() {
FluxMessageChannel outputChannel = new FluxMessageChannel();
StepVerifier verifier =
StepVerifier.create(outputChannel)
.assertNext(m -> {
assertThat(m.getHeaders())
.containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)
.containsEntry(FileHeaders.MARKER, "START");
assertThat(m.getPayload()).isInstanceOf(FileMarker.class);
FileMarker fileMarker = (FileMarker) m.getPayload();
assertThat(fileMarker.getMark()).isEqualTo(FileMarker.Mark.START);
assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath());
})
.expectNextCount(2)
.assertNext(m -> {
assertThat(m.getHeaders()).containsEntry(FileHeaders.MARKER, "END");
assertThat(m.getPayload()).isInstanceOf(FileMarker.class);
FileMarker fileMarker = (FileMarker) m.getPayload();
assertThat(fileMarker.getMark()).isEqualTo(FileMarker.Mark.END);
assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath());
assertThat(fileMarker.getLineCount()).isEqualTo(2);
})
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verifyLater();
FileSplitter splitter = new FileSplitter(true, true);
splitter.setApplySequence(true);
splitter.setOutputChannel(outputChannel);
splitter.handleMessage(new GenericMessage<>(file));
StepVerifier.create(outputChannel)
.assertNext(m -> {
assertThat(m.getHeaders())
.containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)
.containsEntry(FileHeaders.MARKER, "START");
assertThat(m.getPayload()).isInstanceOf(FileSplitter.FileMarker.class);
FileMarker fileMarker = (FileSplitter.FileMarker) m.getPayload();
assertThat(fileMarker.getMark()).isEqualTo(FileMarker.Mark.START);
assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath());
})
.expectNextCount(2)
.assertNext(m -> {
assertThat(m.getHeaders()).containsEntry(FileHeaders.MARKER, "END");
assertThat(m.getPayload()).isInstanceOf(FileSplitter.FileMarker.class);
FileMarker fileMarker = (FileSplitter.FileMarker) m.getPayload();
assertThat(fileMarker.getMark()).isEqualTo(FileMarker.Mark.END);
assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath());
assertThat(fileMarker.getLineCount()).isEqualTo(2);
})
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verify(Duration.ofSeconds(1));
verifier.verify(Duration.ofSeconds(1));
}
@Test