All operators now implement basic Scannable (reactor#1136, root of re…

…actor#1105) Flux and Mono themselves don't implement Scannable because it would make the `scan` operator ambiguous. However, all concrete Flux and Mono operator implementations _should_ implement Scannable, even if very basic (`return null;`), so that things like hooks and lift don't hide the Publisher behind a Scannable.UNAVAILABLE_SCAN. Also improves Hooks javadoc with regards to lift restrictions.
ADelre · Mar 21, 2018 · 9dac17d · 9dac17d
1 parent 5d1372b
commit 9dac17d
Show file tree

Hide file tree

Showing 53 changed files with 423 additions and 62 deletions.
diff --git a/reactor-core/src/main/java/reactor/adapter/JdkFlowAdapter.java b/reactor-core/src/main/java/reactor/adapter/JdkFlowAdapter.java
@@ -22,6 +22,7 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
 import reactor.core.publisher.Flux;
 
 /**
@@ -55,7 +56,7 @@ public static <T> Flux<T> flowPublisherToFlux(Flow.Publisher<T> publisher) {
 		return new FlowPublisherAsFlux<>(publisher);
 	}
 
-	private static class FlowPublisherAsFlux<T> extends Flux<T> {
+	private static class FlowPublisherAsFlux<T> extends Flux<T> implements Scannable {
         private final java.util.concurrent.Flow.Publisher<T> pub;
 
         private FlowPublisherAsFlux(java.util.concurrent.Flow.Publisher<T> pub) {
@@ -66,6 +67,11 @@ private FlowPublisherAsFlux(java.util.concurrent.Flow.Publisher<T> pub) {
         public void subscribe(final CoreSubscriber<? super T> actual) {
         	pub.subscribe(new SubscriberToRS<>(actual));
         }
+
+		@Override
+		public Object scanUnsafe(Attr key) {
+			return null; //no particular key to be represented, still useful in hooks
+		}
     }
 
     private static class PublisherAsFlowPublisher<T> implements Flow.Publisher<T> {

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxArray.java b/reactor-core/src/main/java/reactor/core/publisher/FluxArray.java
@@ -22,6 +22,7 @@
 import org.reactivestreams.Subscriber;
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
+import reactor.core.Scannable;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -31,7 +32,7 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxArray<T> extends Flux<T> implements Fuseable {
+final class FluxArray<T> extends Flux<T> implements Fuseable, Scannable {
 
 	final T[] array;
 
@@ -59,6 +60,13 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		subscribe(actual, array);
 	}
 
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		if (key == Attr.BUFFERED) return array.length;
+		return null;
+	}
+
 	static final class ArraySubscription<T>
 			implements InnerProducer<T>, SynchronousSubscription<T> {
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxCallable.java b/reactor-core/src/main/java/reactor/core/publisher/FluxCallable.java
@@ -20,14 +20,16 @@
 
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
+import reactor.core.Scannable;
 import reactor.util.annotation.Nullable;
 
 /**
  * For each subscriber, a Supplier is invoked and the returned value emitted.
  *
  * @param <T> the value type;
  */
-final class FluxCallable<T> extends Flux<T> implements Callable<T>, Fuseable {
+final class FluxCallable<T> extends Flux<T>
+		implements Callable<T>, Fuseable, Scannable {
 
 	final Callable<T> callable;
 
@@ -59,4 +61,9 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 	public T call() throws Exception {
 		return callable.call();
 	}
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
 }
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxCombineLatest.java b/reactor-core/src/main/java/reactor/core/publisher/FluxCombineLatest.java
@@ -44,7 +44,7 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxCombineLatest<T, R> extends Flux<R> implements Fuseable {
+final class FluxCombineLatest<T, R> extends Flux<R> implements Fuseable, Scannable {
 
 	final Publisher<? extends T>[] array;
 
@@ -176,6 +176,12 @@ public void subscribe(CoreSubscriber<? super R> actual) {
 		coordinator.subscribe(a, n);
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		if (key == Attr.PREFETCH) return prefetch;
+		return null;
+	}
+
 	static final class CombineLatestCoordinator<T, R>
 			implements QueueSubscription<R>, InnerProducer<R> {
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxConcatArray.java b/reactor-core/src/main/java/reactor/core/publisher/FluxConcatArray.java
@@ -22,6 +22,7 @@
 import org.reactivestreams.Publisher;
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
+import reactor.core.Scannable;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -30,7 +31,7 @@
  * @param <T> the value type
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxConcatArray<T> extends Flux<T> {
+final class FluxConcatArray<T> extends Flux<T> implements Scannable {
 
 	final Publisher<? extends T>[] array;
 
@@ -81,6 +82,13 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		}
 	}
 
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		if (key == Attr.DELAY_ERROR) return delayError;
+		return null;
+	}
+
 	/**
 	 * Returns a new instance which has the additional source to be merged together with
 	 * the current array of sources.

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxConcatIterable.java b/reactor-core/src/main/java/reactor/core/publisher/FluxConcatIterable.java
@@ -22,6 +22,7 @@
 
 import org.reactivestreams.Publisher;
 import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
 
 /**
  * Concatenates a fixed array of Publishers' values.
@@ -30,7 +31,7 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxConcatIterable<T> extends Flux<T> {
+final class FluxConcatIterable<T> extends Flux<T> implements Scannable {
 
 	final Iterable<? extends Publisher<? extends T>> iterable;
 
@@ -61,6 +62,11 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		}
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
+
 	static final class ConcatIterableSubscriber<T>
 			extends Operators.MultiSubscriptionSubscriber<T, T> {
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxCreate.java b/reactor-core/src/main/java/reactor/core/publisher/FluxCreate.java
@@ -42,7 +42,7 @@
  *
  * @param <T> the value type
  */
-final class FluxCreate<T> extends Flux<T> {
+final class FluxCreate<T> extends Flux<T> implements Scannable {
 
 	enum CreateMode {
 		PUSH_ONLY, PUSH_PULL
@@ -99,6 +99,11 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		}
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
+
 	/**
 	 * Serializes calls to onNext, onError and onComplete.
 	 *

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxDefer.java b/reactor-core/src/main/java/reactor/core/publisher/FluxDefer.java
@@ -21,6 +21,7 @@
 
 import org.reactivestreams.Publisher;
 import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
 
 /**
  * Defers the creation of the actual Publisher the Subscriber will be subscribed to.
@@ -29,7 +30,7 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxDefer<T> extends Flux<T> {
+final class FluxDefer<T> extends Flux<T> implements Scannable {
 
 	final Supplier<? extends Publisher<? extends T>> supplier;
 
@@ -53,4 +54,10 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 
 		from(p).subscribe(actual);
 	}
+
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
 }
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxEmpty.java b/reactor-core/src/main/java/reactor/core/publisher/FluxEmpty.java
@@ -18,6 +18,7 @@
 
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
+import reactor.core.Scannable;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -27,9 +28,8 @@
  * Use the {@link #instance()} method to obtain a properly type-parametrized view of it.
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxEmpty
-extends Flux<Object>
-		implements Fuseable.ScalarCallable<Object> {
+final class FluxEmpty extends Flux<Object>
+		implements Fuseable.ScalarCallable<Object>, Scannable {
 
 	private static final Flux<Object> INSTANCE = new FluxEmpty();
 
@@ -42,6 +42,11 @@ public void subscribe(CoreSubscriber<? super Object> actual) {
 		Operators.complete(actual);
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
+
 	/**
 	 * Returns a properly parametrized instance of this empty Publisher.
 	 *

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxError.java b/reactor-core/src/main/java/reactor/core/publisher/FluxError.java
@@ -22,6 +22,7 @@
 import reactor.core.Fuseable;
 
 import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
 
 /**
  * Emits a constant or generated Throwable instance to Subscribers.
@@ -30,7 +31,8 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxError<T> extends Flux<T> implements Fuseable.ScalarCallable {
+final class FluxError<T> extends Flux<T>
+		implements Fuseable.ScalarCallable, Scannable {
 
 	final Throwable error;
 
@@ -50,4 +52,9 @@ public Object call() throws Exception {
 		}
 		throw Exceptions.propagate(error);
 	}
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
 }
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxErrorOnRequest.java b/reactor-core/src/main/java/reactor/core/publisher/FluxErrorOnRequest.java
@@ -21,6 +21,7 @@
 
 import org.reactivestreams.Subscriber;
 import reactor.core.CoreSubscriber;
+import reactor.core.Scannable;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -30,7 +31,7 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxErrorOnRequest<T> extends Flux<T> {
+final class FluxErrorOnRequest<T> extends Flux<T> implements Scannable {
 
 	final Throwable error;
 
@@ -43,6 +44,11 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		actual.onSubscribe(new ErrorSubscription(actual, error));
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
+
 	static final class ErrorSubscription implements InnerProducer {
 
 		final CoreSubscriber<?> actual;

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxErrorSupplied.java b/reactor-core/src/main/java/reactor/core/publisher/FluxErrorSupplied.java
@@ -38,6 +38,7 @@
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
 import reactor.core.Fuseable;
+import reactor.core.Scannable;
 
 /**
  * Emits a generated {@link Throwable} instance to Subscribers, lazily generated via a
@@ -47,7 +48,8 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxErrorSupplied<T> extends Flux<T> implements Fuseable.ScalarCallable {
+final class FluxErrorSupplied<T> extends Flux<T>
+		implements Fuseable.ScalarCallable, Scannable {
 
 	final Supplier<Throwable> errorSupplier;
 
@@ -69,4 +71,9 @@ public Object call() throws Exception {
 		}
 		throw Exceptions.propagate(error);
 	}
+
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
 }
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFirstEmitting.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFirstEmitting.java
@@ -35,7 +35,7 @@
  *
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
-final class FluxFirstEmitting<T> extends Flux<T> {
+final class FluxFirstEmitting<T> extends Flux<T> implements Scannable {
 
 	final Publisher<? extends T>[] array;
 
@@ -161,6 +161,11 @@ FluxFirstEmitting<T> ambAdditionalSource(Publisher<? extends T> source) {
 		return null;
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
+
 	static final class RaceCoordinator<T>
 			implements Subscription, Scannable {
 

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxGenerate.java b/reactor-core/src/main/java/reactor/core/publisher/FluxGenerate.java
@@ -24,6 +24,7 @@
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
 import reactor.core.Fuseable;
+import reactor.core.Scannable;
 import reactor.util.annotation.Nullable;
 import reactor.util.context.Context;
 
@@ -38,8 +39,7 @@
  * @param <S> the custom state per subscriber
  * @see <a href="https://github.com/reactor/reactive-streams-commons">https://github.com/reactor/reactive-streams-commons</a>
  */
-final class FluxGenerate<T, S>
-extends Flux<T> implements Fuseable {
+final class FluxGenerate<T, S> extends Flux<T> implements Fuseable, Scannable {
 
 
 	static final Callable EMPTY_CALLABLE = () -> null;
@@ -83,6 +83,11 @@ public void subscribe(CoreSubscriber<? super T> actual) {
 		actual.onSubscribe(new GenerateSubscription<>(actual, state, generator, stateConsumer));
 	}
 
+	@Override
+	public Object scanUnsafe(Attr key) {
+		return null; //no particular key to be represented, still useful in hooks
+	}
+
 	static final class GenerateSubscription<T, S>
 	  implements QueueSubscription<T>, InnerProducer<T>, SynchronousSink<T> {