Fix stress test expectation and bug in ColdTestPublisher (#3700)

This PR fixes an early termination bug in `ColdTestPublisher` which
happens when request from the downstream subscriber races with new
values and following termination which in rare case ends up in
termination without propagation of the remaining values. Also this PR
improves expectation for the stress test where ColdTestPulisher takes a
key role.

closes #3633

---------

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>

reactor-core/src/jcstress/java/reactor/core/publisher/FluxSwitchMapStressTest.java

@@ -272,7 +272,8 @@ public void arbiter(II_Result r) {
 
 	// Ignore, flaky test (https://github.com/reactor/reactor-core/issues/3633)
 	//@JCStressTest
-	@Outcome(id = {"200, 0", "200, 1"}, expect = ACCEPTABLE, desc = "Should produced exactly what was requested")
+	@Outcome(id = {"200, 0, 0", "200, 0, 1"}, expect = ACCEPTABLE, desc = "Should " +
+			"produced exactly what was requested")
 	@State
 	public static class RequestAndProduceStressTest2 extends FluxSwitchMapStressTest {
 
@@ -312,9 +313,17 @@ public void outerRequest() {
 		}
 
 		@Arbiter
-		public void arbiter(II_Result r) {
-			r.r1 = (int) (stressSubscriber.onNextCalls.get() + switchMapMain.requested);
-			r.r2 = stressSubscriber.onCompleteCalls.get();
+		public void arbiter(III_Result r) {
+			r.r1 = stressSubscriber.onNextCalls.get();
+			r.r2 = (int) switchMapMain.requested;
+			r.r3 = stressSubscriber.onCompleteCalls.get();
+
+			switch (r.toString()) {
+				case "200, 0, 0":
+				case "200, 0, 1":
+					break;
+				default: throw new IllegalStateException(r + " " + fastLogger);
+			}
 
 			if (stressSubscriber.onNextCalls.get() < 200 && stressSubscriber.onNextDiscarded.get() < switchMapMain.requested) {
 				throw new IllegalStateException(r + " " + fastLogger);

reactor-test/src/main/java/reactor/test/publisher/ColdTestPublisher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2017-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.
@@ -103,6 +103,7 @@ public void subscribe(Subscriber<? super T> s) {
 		}
 
 		s.onSubscribe(p); // will trigger drain() via request()
+		p.drain(); // ensures that empty source terminal signal is propagated without waiting for a request from the subscriber
 	}
 
 	boolean add(ColdTestPublisherSubscription<T> s) {
@@ -315,7 +316,7 @@ private void drain() {
 		 * @return true if the TestPublisher was terminated, false otherwise
 		 */
 		private boolean emitTerminalSignalIfAny() {
-			if (parent.done) {
+			if (parent.done && this.parent.values.size() == index) {
 				parent.remove(this);
 
 				final Throwable t = parent.error;