Merge #3156 into 3.5.0-M6

Signed-off-by: Oleh Dokuka <odokuka@vmware.com> Signed-off-by: OlegDokuka <odokuka@vmware.com>
reactor · Aug 16, 2022 · 8a31886 · 8a31886
2 parents c747f80 + 7f6f65f
commit 8a31886
Show file tree

Hide file tree

Showing 3 changed files with 36 additions and 2 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/Flux.java b/reactor-core/src/main/java/reactor/core/publisher/Flux.java
@@ -8438,6 +8438,10 @@ public final void subscribe(Subscriber<? super T> actual) {
 		CorePublisher publisher = Operators.onLastAssembly(this);
 		CoreSubscriber subscriber = Operators.toCoreSubscriber(actual);
 
+		if (subscriber instanceof Fuseable.QueueSubscription && this != publisher && this instanceof Fuseable && !(publisher instanceof Fuseable)) {
+			subscriber = new FluxHide.SuppressFuseableSubscriber<>(subscriber);
+		}
+
 		try {
 			if (publisher instanceof OptimizableOperator) {
 				OptimizableOperator operator = (OptimizableOperator) publisher;

diff --git a/reactor-core/src/main/java/reactor/core/publisher/Mono.java b/reactor-core/src/main/java/reactor/core/publisher/Mono.java
@@ -4299,6 +4299,10 @@ public final void subscribe(Subscriber<? super T> actual) {
 		CorePublisher publisher = Operators.onLastAssembly(this);
 		CoreSubscriber subscriber = Operators.toCoreSubscriber(actual);
 
+		if (subscriber instanceof Fuseable.QueueSubscription && this != publisher && this instanceof Fuseable && !(publisher instanceof Fuseable)) {
+			subscriber = new FluxHide.SuppressFuseableSubscriber<>(subscriber);
+		}
+
 		try {
 			if (publisher instanceof OptimizableOperator) {
 				OptimizableOperator operator = (OptimizableOperator) publisher;

diff --git a/reactor-core/src/test/java/reactor/core/publisher/HooksTest.java b/reactor-core/src/test/java/reactor/core/publisher/HooksTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2022 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.
@@ -70,7 +70,6 @@ public TestException(String message) {
 		}
 	}
 
-
 	@Test
 	public void staticActivationOfOperatorDebug() {
 		String oldProp = System.setProperty("reactor.trace.operatorStacktrace", "true");
@@ -142,6 +141,7 @@ public void onEachOperatorOneHookNoComposite() {
 		Function<? super Publisher<Object>, ? extends Publisher<Object>> hook = p -> p;
 		Hooks.onEachOperator(hook);
 
+
 		assertThat(Hooks.onEachOperatorHook).isSameAs(hook);
 	}
 
@@ -1254,4 +1254,30 @@ public void onComplete() {
 			}
 		};
 	}
+
+	// https://github.com/reactor/reactor-core/issues/3137
+	@Test
+	public void reproduceClassCastExceptionWithHooks() {
+		Hooks.onLastOperator(objectPublisher -> {
+			if (objectPublisher instanceof Mono) {
+				return Hooks.convertToMonoBypassingHooks(objectPublisher, false)
+				            .doFinally(signalType -> {
+				            });
+			} else {
+				return objectPublisher;
+			}
+		});
+
+		try {
+			Mono.just(1)
+			    .flatMap(fsm -> Mono.just(1)
+			                        .doOnSubscribe(subscription -> {
+			                        }))
+			    .doOnSubscribe(subscription -> {
+			    })
+			    .block();
+		} finally {
+			Hooks.resetOnLastOperator();
+		}
+	}
 }