Make some Mono sources and aggregators lazier

reactor-core/src/main/java/reactor/core/publisher/FluxDefaultIfEmpty.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.
@@ -17,6 +17,8 @@
 package reactor.core.publisher;
 
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
@@ -50,37 +52,80 @@ public Object scanUnsafe(Attr key) {
 	}
 
 	static final class DefaultIfEmptySubscriber<T>
-			extends Operators.MonoSubscriber<T, T> {
+			implements InnerOperator<T, T>,
+			           Fuseable, //for constants only
+			           Fuseable.QueueSubscription<T> {
+
+		final CoreSubscriber<? super T> actual;
 
 		Subscription s;
 
+		boolean hasRequest;
+
 		boolean hasValue;
 
-		DefaultIfEmptySubscriber(CoreSubscriber<? super T> actual, T value) {
-			super(actual);
-			//noinspection deprecation
-			this.value = value; //we write once, setValue() is NO-OP
+		volatile T fallbackValue;
+		@SuppressWarnings("rawtypes")
+		static final AtomicReferenceFieldUpdater<DefaultIfEmptySubscriber, Object> FALLBACK_VALUE =
+				AtomicReferenceFieldUpdater.newUpdater(DefaultIfEmptySubscriber.class, Object.class, "fallbackValue");
+
+		volatile int state;
+		@SuppressWarnings("rawtypes")
+		static final AtomicIntegerFieldUpdater<DefaultIfEmptySubscriber> STATE =
+				AtomicIntegerFieldUpdater.newUpdater(DefaultIfEmptySubscriber.class, "state");
+
+		DefaultIfEmptySubscriber(CoreSubscriber<? super T> actual, T fallbackValue) {
+			this.actual = actual;
+			FALLBACK_VALUE.lazySet(this, fallbackValue);
+		}
+
+		@Override
+		public CoreSubscriber<? super T> actual() {
+			return this.actual;
 		}
 
 		@Override
 		@Nullable
 		public Object scanUnsafe(Attr key) {
 			if (key == Attr.PARENT) return s;
+			if (key == Attr.PREFETCH) return 0;
 			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
 
-			return super.scanUnsafe(key);
+			return InnerOperator.super.scanUnsafe(key);
 		}
 
 		@Override
 		public void request(long n) {
-			super.request(n);
+			if (!hasRequest) {
+				hasRequest = true;
+
+				final int state = this.state;
+
+				if (state != 1 && STATE.compareAndSet(this, state, state | 1)) {
+					if (state > 1) {
+						final T fallbackValue = this.fallbackValue;
+						if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this,
+								fallbackValue,
+								null)) {
+							// completed before request means source was empty
+							actual.onNext(fallbackValue);
+							actual.onComplete();
+ 						}
+						return;
+					}
+				}
+			}
+
 			s.request(n);
 		}
 
 		@Override
 		public void cancel() {
-			super.cancel();
 			s.cancel();
+			final T fallbackValue = this.fallbackValue;
+			if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this, fallbackValue, null)) {
+				Operators.onDiscard(fallbackValue, actual.currentContext());
+			}
 		}
 
 		@Override
@@ -96,29 +141,84 @@ public void onSubscribe(Subscription s) {
 		public void onNext(T t) {
 			if (!hasValue) {
 				hasValue = true;
+
+				final T fallbackValue = this.fallbackValue;
+				if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this, fallbackValue, null)) {
+					Operators.onDiscard(fallbackValue, actual.currentContext());
+				}
 			}
 
 			actual.onNext(t);
 		}
 
 		@Override
 		public void onComplete() {
-			if (hasValue) {
-				actual.onComplete();
-			} else {
-				complete(this.value);
+			if (!hasValue) {
+				if (hasRequest) {
+					final T fallbackValue = this.fallbackValue;
+					if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this,
+							fallbackValue,
+							null)) {
+						actual.onNext(fallbackValue);
+						actual.onComplete();
+					}
+					return;
+				}
+
+				final int state = this.state;
+				if (state == 0 && STATE.compareAndSet(this, 0, 2)) {
+					return;
+				}
+
+				final T fallbackValue = this.fallbackValue;
+				if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this,
+						fallbackValue,
+						null)) {
+					actual.onNext(fallbackValue);
+					actual.onComplete();
+				}
+
+				return;
 			}
+
+			actual.onComplete();
 		}
 
 		@Override
-		public void setValue(T value) {
-			// value is constant. writes from the base class are redundant, and the constant
-			// would always be visible in cancel(), so it will safely be discarded.
+		public void onError(Throwable t) {
+			if (!hasValue) {
+				final T fallbackValue = this.fallbackValue;
+				if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this, fallbackValue, null)) {
+					Operators.onDiscard(t, actual.currentContext());
+				}
+			}
+
+			actual.onError(t);
 		}
 
 		@Override
 		public int requestFusion(int requestedMode) {
 			return Fuseable.NONE; // prevent fusion because of the upstream
 		}
+
+		@Override
+		public T poll() {
+			return null;
+		}
+
+		@Override
+		public int size() {
+			return 0;
+		}
+
+		@Override
+		public boolean isEmpty() {
+			return false;
+		}
+
+		@Override
+		public void clear() {
+
+		}
 	}
 }

reactor-core/src/main/java/reactor/core/publisher/MonoAll.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.
@@ -17,6 +17,7 @@
 package reactor.core.publisher;
 
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.Predicate;
 
 import org.reactivestreams.Subscription;
@@ -55,48 +56,89 @@ public Object scanUnsafe(Attr key) {
 		return super.scanUnsafe(key);
 	}
 
-	static final class AllSubscriber<T> extends Operators.MonoSubscriber<T, Boolean> {
+	static final class AllSubscriber<T> implements InnerOperator<T, Boolean>,
+	                                               Fuseable, //for constants only
+	                                               QueueSubscription<Boolean> {
 		final Predicate<? super T> predicate;
+		final CoreSubscriber<? super Boolean> actual;
 
 		Subscription s;
 
+		boolean hasRequest;
+
 		boolean done;
 
+		volatile int state;
+		@SuppressWarnings("rawtypes")
+		static final AtomicIntegerFieldUpdater<AllSubscriber> STATE =
+				AtomicIntegerFieldUpdater.newUpdater(AllSubscriber.class, "state");
+
 		AllSubscriber(CoreSubscriber<? super Boolean> actual, Predicate<? super T> predicate) {
-			super(actual);
+			this.actual = actual;
 			this.predicate = predicate;
 		}
 
+		@Override
+		public int requestFusion(int requestedMode) {
+			return Fuseable.NONE;
+		}
+
+		@Override
+		public CoreSubscriber<? super Boolean> actual() {
+			return this.actual;
+		}
+
 		@Override
 		@Nullable
 		public Object scanUnsafe(Attr key) {
 			if (key == Attr.TERMINATED) return done;
 			if (key == Attr.PARENT) return s;
 			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+			if (key == Attr.PREFETCH) return Integer.MAX_VALUE;
 
-			return super.scanUnsafe(key);
+			return InnerOperator.super.scanUnsafe(key);
+		}
+
+		@Override
+		public void request(long n) {
+			if (!hasRequest) {
+				hasRequest = true;
+
+				final int state = this.state;
+				if ((state & 1) == 1) {
+					return;
+				}
+
+				if (STATE.compareAndSet(this, state, state | 1)) {
+					if (state == 0) {
+						s.request(Long.MAX_VALUE);
+					}
+					else {
+						// completed before request means source was empty
+						actual.onNext(true);
+						actual.onComplete();
+					}
+				}
+			}
 		}
 
 		@Override
 		public void cancel() {
 			s.cancel();
-			super.cancel();
 		}
 
 		@Override
 		public void onSubscribe(Subscription s) {
 			if (Operators.validate(this.s, s)) {
 				this.s = s;
 				actual.onSubscribe(this);
-
-				s.request(Long.MAX_VALUE);
 			}
 		}
 
 		@Override
 		public void onNext(T t) {
-
 			if (done) {
+				Operators.onDiscard(t, this.actual.currentContext());
 				return;
 			}
 
@@ -113,7 +155,8 @@ public void onNext(T t) {
 				done = true;
 				s.cancel();
 
-				complete(false);
+				this.actual.onNext(false);
+				this.actual.onComplete();
 			}
 		}
 
@@ -134,8 +177,40 @@ public void onComplete() {
 				return;
 			}
 			done = true;
-			complete(true);
+
+			if (hasRequest) {
+				actual.onNext(true);
+				actual.onComplete();
+				return;
+			}
+
+			final int state = this.state;
+			if (state == 0 && STATE.compareAndSet(this, 0, 2)) {
+				return;
+			}
+
+			actual.onNext(true);
+			actual.onComplete();
+		}
+
+		@Override
+		public Boolean poll() {
+			return null;
+		}
+
+		@Override
+		public int size() {
+			return 0;
+		}
+
+		@Override
+		public boolean isEmpty() {
+			return false;
 		}
 
+		@Override
+		public void clear() {
+
+		}
 	}
 }