Fix SwitchOnFirst hanging subscription (#3939)

In a situation where `SwitchOnFirstMain` is subscribed to after the operator itself has terminated, the expectation is that either an error is propagated due to cancellation of the source or the first actual subscription succeeds. Due to a regression introduced in #2794 the subscription would hang with no termination. This change addresses the issue and the subscrption properly terminates. Related to #3936. Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
reactor · Nov 21, 2024 · a7fe500 · a7fe500
1 parent aea75c2
commit a7fe500
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 3 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchOnFirst.java b/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchOnFirst.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2023 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-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.
@@ -716,7 +716,7 @@ public final void subscribe(CoreSubscriber<? super T> inboundSubscriber) {
 				return;
 			}
 
-			if (hasInboundClosedPrematurely(previousState)) {
+			if (hasInboundCancelled(previousState) ||hasInboundClosedPrematurely(previousState)) {
 				Operators.error(inboundSubscriber, new CancellationException("FluxSwitchOnFirst has already been cancelled"));
 				return;
 			}

diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxSwitchOnFirstTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxSwitchOnFirstTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-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.
@@ -45,6 +45,7 @@
 import reactor.core.scheduler.Schedulers;
 import reactor.test.MockUtils;
 import reactor.test.StepVerifier;
+import reactor.test.StepVerifierOptions;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.scheduler.VirtualTimeScheduler;
 import reactor.test.util.RaceTestUtils;
@@ -1386,6 +1387,35 @@ public void shouldCancelSourceOnOnDownstreamTerminal() {
         assertThat(testPublisher.wasCancelled()).isTrue();
     }
 
+	@Test
+	void shouldErrorWhenResubscribingAfterCancel() {
+		TestPublisher<Long> testPublisher = TestPublisher.create();
+
+		AtomicReference<Flux<Long>> intercepted = new AtomicReference<>();
+
+		Flux<Flux<Long>> flux =
+				testPublisher.flux().switchOnFirst((s, f) -> {
+					intercepted.set(f);
+					// Due to wrapping we can avoid subscribing to f and subscribe later
+					// (re-subscribing is not allowed)
+					return Mono.just(f);
+				});
+
+		StepVerifier.create(flux)
+		            .expectSubscription()
+		            .then(() -> testPublisher.next(1L))
+					.thenCancel()
+		            .verify(Duration.ofSeconds(2));
+
+		Flux<Long> switchOnFirstMain = intercepted.get();
+
+		StepVerifier.create(switchOnFirstMain,
+				            StepVerifierOptions.create()
+				                               .scenarioName("Expect immediate onError from SwitchOnFirstMain as the outer operator has completed"))
+		            .expectError(IllegalStateException.class)
+		            .verify(Duration.ofSeconds(1));
+	}
+
     @Test
     public void scanOperator(){
     	Flux<Integer> parent = Flux.just(1);