Merge reactor#1982 into 3.3

linagora · Dec 5, 2019 · 1628b2e · 1628b2e
2 parents f614b1a + ce40bd0
commit 1628b2e
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 1 deletion.
diff --git a/reactor-test/src/main/java/reactor/test/DefaultStepVerifierBuilder.java b/reactor-test/src/main/java/reactor/test/DefaultStepVerifierBuilder.java
@@ -1645,8 +1645,12 @@ void waitTaskEvent() {
 				try {
 					if (event instanceof SubscriptionTaskEvent) {
 						updateRequested(event);
+						((TaskEvent<T>) event).run(this);
+						serializeDrainAndSubscriptionEvent();
+					}
+					else {
+						((TaskEvent<T>) event).run(this);
 					}
-					((TaskEvent<T>) event).run(this);
 				}
 				catch (Throwable t) {
 					Exceptions.throwIfFatal(t);

diff --git a/reactor-test/src/test/java/reactor/test/StepVerifierTests.java b/reactor-test/src/test/java/reactor/test/StepVerifierTests.java
@@ -38,6 +38,7 @@
 import org.junit.Test;
 import reactor.core.Fuseable;
 import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.UnicastProcessor;
@@ -2312,4 +2313,34 @@ public void verifyLaterCanVerifyConnectableFlux_withAssertionErrors() {
 				.isThrownBy(() -> deferred2.verify(Duration.ofSeconds(10)))
 				.withMessage("expectation \"expectNext(5)\" failed (expected value: 5; actual value: 3)");
 	}
+
+	@Test
+	public void verifyDrainOnRequestInCaseOfFusion() {
+		MonoProcessor<Integer> processor = MonoProcessor.create();
+		StepVerifier.create(processor, 0)
+				.expectFusion(Fuseable.ANY)
+				.then(() -> processor.onNext(1))
+				.thenRequest(1)
+				.expectNext(1)
+				.verifyComplete();
+	}
+
+	@Test
+	public void verifyDrainOnRequestInCaseOfFusion2() {
+		ArrayList<Long> requests = new ArrayList<>();
+		UnicastProcessor<Integer> processor = UnicastProcessor.create();
+		StepVerifier.create(processor.doOnRequest(requests::add), 0)
+				.expectFusion(Fuseable.ANY)
+				.then(() -> {
+					processor.onNext(1);
+					processor.onComplete();
+				})
+				.thenRequest(1)
+				.thenRequest(1)
+				.thenRequest(1)
+				.expectNext(1)
+				.verifyComplete();
+
+		assertThat(requests).containsExactly(1L, 1L, 1L);
+	}
 }