Restore where onLastOperatorHook is applied (#3673)

Recent improvements in the automatic context propagation (#3549) resulted in a regression – some sites where the "last operator" hook was previously applied no longer saw that behaviour. This change restores it. Implementation wise, it's worth noting that the "last operator" functionality relies on executing the subscribe(Subscriber) method from the base reactive-streams Publisher instead of the overloads that come from CorePublisher. The implementations of the reactive-streams base method in reactor-core apply this hook and that is when something is considered a "last operator". The wrapping of the Publisher when a non-internal producer is encountered to restore ThreadLocal values has changed the compiler's inference of the signature to use the CoreSubscriber argument variant, breaking the behaviour. This commit does not bring any tests as the functionality was not extensively tested before. The issue was discovered in spring-security in spring-projects/spring-security#14207 and the change has been validated against the actual use case.
reactor · Jan 2, 2024 · 378543d · 378543d
1 parent 4e6f40e
commit 378543d
Show file tree

Hide file tree

Showing 24 changed files with 85 additions and 52 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxBufferWhen.java b/reactor-core/src/main/java/reactor/core/publisher/FluxBufferWhen.java
@@ -360,7 +360,8 @@ void open(OPEN token) {
 
 			BufferWhenCloseSubscriber<T, BUFFER> bc = new BufferWhenCloseSubscriber<>(this, idx);
 			subscribers.add(bc);
-			Operators.toFluxOrMono(p).subscribe(bc);
+			p = Operators.toFluxOrMono(p);
+			p.subscribe(bc);
 		}
 
 		void openComplete(BufferWhenOpenSubscriber<OPEN> os) {

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxConcatArray.java b/reactor-core/src/main/java/reactor/core/publisher/FluxConcatArray.java
@@ -61,7 +61,8 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 			if (p == null) {
 				Operators.error(actual, new NullPointerException("The single source Publisher is null"));
 			} else {
-				Operators.toFluxOrMono(p).subscribe(actual);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(actual);
 			}
 			return;
 		}
@@ -255,7 +256,8 @@ public void onComplete() {
 				if (this.cancelled) {
 					return;
 				}
-				Operators.toFluxOrMono(p).subscribe(this);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(this);
 
 				final Object state = this.get();
 				if (state != DONE) {
@@ -404,7 +406,7 @@ public void onComplete() {
 					return;
 				}
 
-				final Publisher<? extends T> p = a[i];
+				Publisher<? extends T> p = a[i];
 
 				if (p == null) {
 					this.remove();
@@ -440,7 +442,8 @@ public void onComplete() {
 					return;
 				}
 
-				Operators.toFluxOrMono(p).subscribe(this);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(this);
 
 				final Object state = this.get();
 				if (state != DONE) {

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxConcatIterable.java b/reactor-core/src/main/java/reactor/core/publisher/FluxConcatIterable.java
@@ -144,7 +144,8 @@ public void onComplete() {
 						produced(c);
 					}
 
-					Operators.toFluxOrMono(p).subscribe(this);
+					p = Operators.toFluxOrMono(p);
+					p.subscribe(this);
 
 					if (isCancelled()) {
 						return;

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxConcatMap.java b/reactor-core/src/main/java/reactor/core/publisher/FluxConcatMap.java
@@ -448,7 +448,8 @@ void drain() {
 							}
 							else {
 								active = true;
-								Operators.toFluxOrMono(p).subscribe(inner);
+								p = Operators.toFluxOrMono(p);
+								p.subscribe(inner);
 							}
 						}
 					}
@@ -805,7 +806,8 @@ void drain() {
 							}
 							else {
 								active = true;
-								Operators.toFluxOrMono(p).subscribe(inner);
+								p = Operators.toFluxOrMono(p);
+								p.subscribe(inner);
 							}
 						}
 					}

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxConcatMapNoPrefetch.java b/reactor-core/src/main/java/reactor/core/publisher/FluxConcatMapNoPrefetch.java
@@ -203,7 +203,8 @@ public void onNext(T t) {
 					return;
 				}
 
-				Operators.toFluxOrMono(p).subscribe(inner);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(inner);
 			}
 			catch (Throwable e) {
 				Context ctx = actual.currentContext();

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFilterWhen.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFilterWhen.java
@@ -279,7 +279,8 @@ void drain() {
 								FilterWhenInner inner = new FilterWhenInner(this, !(p instanceof Mono));
 								if (CURRENT.compareAndSet(this,null, inner)) {
 									state = STATE_RUNNING;
-									Operators.toFluxOrMono(p).subscribe(inner);
+									p = Operators.toFluxOrMono(p);
+									p.subscribe(inner);
 									break;
 								}
 							}

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFirstWithSignal.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFirstWithSignal.java
@@ -127,7 +127,8 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 						new NullPointerException("The single source Publisher is null"));
 			}
 			else {
-				Operators.toFluxOrMono(p).subscribe(actual);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(actual);
 			}
 			return;
 		}

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFirstWithValue.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFirstWithValue.java
@@ -227,6 +227,8 @@ void subscribe(Publisher<? extends T>[] sources,
 
 			actual.onSubscribe(this);
 
+			Operators.toFluxOrMono(sources);
+
 			for (int i = 0; i < n; i++) {
 				if (cancelled || winner != Integer.MIN_VALUE) {
 					return;
@@ -237,7 +239,7 @@ void subscribe(Publisher<? extends T>[] sources,
 					return;
 				}
 
-				Operators.toFluxOrMono(sources[i]).subscribe(subscribers[i]);
+				sources[i].subscribe(subscribers[i]);
 			}
 		}
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxGroupJoin.java b/reactor-core/src/main/java/reactor/core/publisher/FluxGroupJoin.java
@@ -336,7 +336,8 @@ void drain() {
 								new LeftRightEndSubscriber(this, true, idx);
 						cancellations.add(end);
 
-						Operators.toFluxOrMono(p).subscribe(end);
+						p = Operators.toFluxOrMono(p);
+						p.subscribe(end);
 
 						ex = error;
 						if (ex != null) {
@@ -404,7 +405,8 @@ else if (mode == RIGHT_VALUE) {
 								new LeftRightEndSubscriber(this, false, idx);
 						cancellations.add(end);
 
-						Operators.toFluxOrMono(p).subscribe(end);
+						p = Operators.toFluxOrMono(p);
+						p.subscribe(end);
 
 						ex = error;
 						if (ex != null) {

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxJoin.java b/reactor-core/src/main/java/reactor/core/publisher/FluxJoin.java
@@ -26,7 +26,6 @@
 import java.util.function.BiFunction;
 import java.util.function.BiPredicate;
 import java.util.function.Function;
-import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 import org.reactivestreams.Publisher;
@@ -289,7 +288,8 @@ void drain() {
 								new LeftRightEndSubscriber(this, true, idx);
 						cancellations.add(end);
 
-						Operators.toFluxOrMono(p).subscribe(end);
+						p = Operators.toFluxOrMono(p);
+						p.subscribe(end);
 
 						ex = error;
 						if (ex != null) {
@@ -366,7 +366,8 @@ else if (mode == RIGHT_VALUE) {
 								new LeftRightEndSubscriber(this, false, idx);
 						cancellations.add(end);
 
-						Operators.toFluxOrMono(p).subscribe(end);
+						p = Operators.toFluxOrMono(p);
+						p.subscribe(end);
 
 						ex = error;
 						if (ex != null) {

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxRepeatWhen.java b/reactor-core/src/main/java/reactor/core/publisher/FluxRepeatWhen.java
@@ -76,7 +76,8 @@ public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> act
 			return null;
 		}
 
-		Operators.toFluxOrMono(p).subscribe(other);
+		p = Operators.toFluxOrMono(p);
+		p.subscribe(other);
 
 		if (!main.cancelled) {
 			return main;

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxRetryWhen.java b/reactor-core/src/main/java/reactor/core/publisher/FluxRetryWhen.java
@@ -74,7 +74,8 @@ static <T> void subscribe(CoreSubscriber<? super T> s,
 			return;
 		}
 
-		Operators.toFluxOrMono(p).subscribe(other);
+		p = Operators.toFluxOrMono(p);
+		p.subscribe(other);
 
 		if (!main.cancelled) {
 			wrapped.subscribe(main);

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxSampleFirst.java b/reactor-core/src/main/java/reactor/core/publisher/FluxSampleFirst.java
@@ -191,7 +191,8 @@ public void onNext(T t) {
 				SampleFirstOther<U> other = new SampleFirstOther<>(this);
 
 				if (Operators.replace(OTHER, this, other)) {
-					Operators.toFluxOrMono(p).subscribe(other);
+					p = Operators.toFluxOrMono(p);
+					p.subscribe(other);
 				}
 			}
 			else {

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxSampleTimeout.java b/reactor-core/src/main/java/reactor/core/publisher/FluxSampleTimeout.java
@@ -207,7 +207,8 @@ public void onNext(T t) {
 			SampleTimeoutOther<T, U> os = new SampleTimeoutOther<>(this, t, idx);
 
 			if (Operators.replace(OTHER, this, os)) {
-				Operators.toFluxOrMono(p).subscribe(os);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(os);
 			}
 		}
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchMap.java b/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchMap.java
@@ -233,7 +233,8 @@ public void onNext(T t) {
 
 			if (INNER.compareAndSet(this, si, innerSubscriber)) {
 				ACTIVE.getAndIncrement(this);
-				Operators.toFluxOrMono(p).subscribe(innerSubscriber);
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(innerSubscriber);
 			}
 		}
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchMapNoPrefetch.java b/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchMapNoPrefetch.java
@@ -215,7 +215,8 @@ void subscribeInner(T nextElement, SwitchMapInner<T, R> nextInner, int nextIndex
 				return;
 			}
 
-			Operators.toFluxOrMono(p).subscribe(nextInner);
+			p = Operators.toFluxOrMono(p);
+			p.subscribe(nextInner);
 		}
 
 		@Override

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchOnFirst.java b/reactor-core/src/main/java/reactor/core/publisher/FluxSwitchOnFirst.java
@@ -519,7 +519,7 @@ public final void onNext(T t) {
 					return;
 				}
 
-				final Publisher<? extends R> outboundPublisher;
+				Publisher<? extends R> outboundPublisher;
 				final SwitchOnFirstControlSubscriber<? super R> o = this.outboundSubscriber;
 
 				try {
@@ -542,7 +542,8 @@ public final void onNext(T t) {
 					return;
 				}
 
-				Operators.toFluxOrMono(outboundPublisher).subscribe(o);
+				outboundPublisher = Operators.toFluxOrMono(outboundPublisher);
+				outboundPublisher.subscribe(o);
 				return;
 			}
 
@@ -575,7 +576,7 @@ public final void onError(Throwable t) {
 			}
 
 			if (!hasFirstValueReceived(previousState)) {
-				final Publisher<? extends R> result;
+				Publisher<? extends R> result;
 				final CoreSubscriber<? super R> o = this.outboundSubscriber;
 				try {
 					final Signal<T> signal = Signal.error(t, o.currentContext());
@@ -586,7 +587,8 @@ public final void onError(Throwable t) {
 					return;
 				}
 
-				Operators.toFluxOrMono(result).subscribe(o);
+				result = Operators.toFluxOrMono(result);
+				result.subscribe(o);
 			}
 		}
 
@@ -611,7 +613,7 @@ public final void onComplete() {
 			}
 
 			if (!hasFirstValueReceived(previousState)) {
-				final Publisher<? extends R> result;
+				Publisher<? extends R> result;
 				final CoreSubscriber<? super R> o = this.outboundSubscriber;
 
 				try {
@@ -623,7 +625,8 @@ public final void onComplete() {
 					return;
 				}
 
-				Operators.toFluxOrMono(result).subscribe(o);
+				result = Operators.toFluxOrMono(result);
+				result.subscribe(o);
 			}
 		}
 
@@ -844,7 +847,7 @@ public boolean tryOnNext(T t) {
 					return true;
 				}
 
-				final Publisher<? extends R> result;
+				Publisher<? extends R> result;
 				final SwitchOnFirstControlSubscriber<? super R> o = this.outboundSubscriber;
 
 				try {
@@ -868,7 +871,8 @@ public boolean tryOnNext(T t) {
 					return true;
 				}
 
-				Operators.toFluxOrMono(result).subscribe(o);
+				result = Operators.toFluxOrMono(result);
+				result.subscribe(o);
 				return true;
 			}
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxUsingWhen.java b/reactor-core/src/main/java/reactor/core/publisher/FluxUsingWhen.java
@@ -19,7 +19,6 @@
 import java.util.Objects;
 import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 
@@ -91,7 +90,8 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 							asyncCancel,
 							null);
 
-					Operators.toFluxOrMono(p).subscribe(subscriber);
+					p = Operators.toFluxOrMono(p);
+					p.subscribe(subscriber);
 				}
 			}
 			catch (Throwable e) {
@@ -101,8 +101,9 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		}
 
 		//trigger the resource creation and delay the subscription to actual
-		Operators.toFluxOrMono(resourceSupplier).subscribe(new ResourceSubscriber(actual,
-				resourceClosure,	asyncComplete, asyncError, asyncCancel, resourceSupplier instanceof Mono));
+		// + ensure onLastOperatorHook is called by invoking Publisher::subscribe(Subscriber)
+		((Publisher<?>) Operators.toFluxOrMono(resourceSupplier)).subscribe(
+				new ResourceSubscriber(actual, resourceClosure,	asyncComplete, asyncError, asyncCancel, resourceSupplier instanceof Mono));
 	}
 
 	@Override
@@ -191,9 +192,10 @@ public void onNext(S resource) {
 			}
 			resourceProvided = true;
 
-			final Publisher<? extends T> p = deriveFluxFromResource(resource, resourceClosure);
+			Publisher<? extends T> p = deriveFluxFromResource(resource, resourceClosure);
 
-			Operators.toFluxOrMono(p).subscribe(FluxUsingWhen.<S, T>prepareSubscriberForResource(resource,
+			p = Operators.toFluxOrMono(p);
+			p.subscribe(FluxUsingWhen.<S, T>prepareSubscriberForResource(resource,
 					this.actual,
 					this.asyncComplete,
 					this.asyncError,
@@ -362,7 +364,8 @@ public void onError(Throwable t) {
 					return;
 				}
 
-				Operators.toFluxOrMono(p).subscribe(new RollbackInner(this, t));
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(new RollbackInner(this, t));
 			}
 		}
 
@@ -382,7 +385,8 @@ public void onComplete() {
 					return;
 				}
 
-				Operators.toFluxOrMono(p).subscribe(new CommitInner(this));
+				p = Operators.toFluxOrMono(p);
+				p.subscribe(new CommitInner(this));
 			}
 		}
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/MonoDelayUntil.java b/reactor-core/src/main/java/reactor/core/publisher/MonoDelayUntil.java
@@ -287,7 +287,7 @@ void subscribeNextTrigger() {
 			final Function<? super T, ? extends Publisher<?>> generator =
 					this.otherGenerators[this.index];
 
-			final Publisher<?> p;
+			Publisher<?> p;
 
 			try {
 				p = generator.apply(this.value);
@@ -304,7 +304,8 @@ void subscribeNextTrigger() {
 				this.triggerSubscriber = triggerSubscriber;
 			}
 
-			Operators.toFluxOrMono(p).subscribe(triggerSubscriber);
+			p = Operators.toFluxOrMono(p);
+			p.subscribe(triggerSubscriber);
 		}
 
 		@Override