Re-work bufferTimeout with fair backpressure to consistent state machine

reactor · Apr 29, 2024 · 16d2d1f · 16d2d1f
1 parent 7f35d1d
commit 16d2d1f
Show file tree

Hide file tree

Showing 5 changed files with 445 additions and 306 deletions.
diff --git a/reactor-core/src/jcstress/java/reactor/core/publisher/FluxBufferTimeoutStressTest.java b/reactor-core/src/jcstress/java/reactor/core/publisher/FluxBufferTimeoutStressTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2023 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2023-2024 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -98,6 +98,9 @@ public void arbiter(LL_Result result) {
 	@Outcome(id = "3, 1", expect = Expect.ACCEPTABLE, desc = "")
 	@Outcome(id = "4, 1", expect = Expect.ACCEPTABLE, desc = "")
 	@Outcome(id = "5, 1", expect = Expect.ACCEPTABLE, desc = "")
+	@Outcome(id = "3, 2", expect = Expect.ACCEPTABLE, desc = "")
+	@Outcome(id = "4, 2", expect = Expect.ACCEPTABLE, desc = "")
+	@Outcome(id = "5, 2", expect = Expect.ACCEPTABLE, desc = "")
 	@State
 	public static class FluxBufferTimeoutStressTestRaceDeliveryAndMoreTimeouts {
 
@@ -169,6 +172,7 @@ public void arbiter(LL_Result result) {
 	@Outcome(id = "5, 0, 3", expect = Expect.ACCEPTABLE, desc = "")
 	@Outcome(id = "5, 0, 4", expect = Expect.ACCEPTABLE, desc = "")
 	@Outcome(id = "5, 0, 5", expect = Expect.ACCEPTABLE, desc = "")
+	@Outcome(id = "5, 0, 1", expect = Expect.ACCEPTABLE, desc = "")
 	@State
 	public static class FluxBufferTimeoutStressTestRaceDeliveryAndMoreTimeoutsPossiblyIncomplete {
 
@@ -242,8 +246,7 @@ public void arbiter(LLL_Result result) {
 				}
 			}
 
-			// TODO #onNext < 5 and incomplete => why fail?
-			if (subscriber.onNextCalls.get() < 5 && subscriber.onCompleteCalls.get() == 0) {
+			if (subscriber.onNextCalls.get() < 5) {
 				fail(fastLogger, "incomplete. received: " + subscriber.receivedValues + "; requested=" + requested.get() + "; result=" + result);
 			}