Introduce Mono.delayElement, prepare move to Flux.delayElements

src/main/java/reactor/core/publisher/Flux.java

@@ -2837,48 +2837,96 @@ public final Flux<T> defaultIfEmpty(T defaultV) {
 	}
 
 	/**
-	 * Delay this {@link Flux} signals to {@link Subscriber#onNext} until the given period elapses.
+	 * Delay each of this {@link Flux} elements ({@link Subscriber#onNext} signals)
+	 * by a given duration.
 	 *
 	 * <p>
 	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
 	 *
-	 * @param delay duration to delay each {@link Subscriber#onNext} call
-	 *
-	 * @return a throttled {@link Flux}
-	 *
+	 * @param delay duration to delay each {@link Subscriber#onNext} signal
+	 * @return a delayed {@link Flux}
+	 * @deprecated will be replaced by {@link #delayElements(Duration)} in 3.1.0
+	 * @see #delaySubscription(Duration) delaySubscription to introduce a delay at the beginning of the sequence only
 	 */
+	@Deprecated
 	public final Flux<T> delay(Duration delay) {
-		return delayMillis(delay.toMillis());
+		return delayElements(delay);
 	}
 
 	/**
-	 * Delay this {@link Flux} signals to {@link Subscriber#onNext} until the given period in milliseconds elapses.
+	 * Delay each of this {@link Flux} elements ({@link Subscriber#onNext} signals)
+	 * by a given duration.
 	 *
 	 * <p>
 	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
 	 *
-	 * @param delay period to delay each {@link Subscriber#onNext} call in milliseconds
+	 * @param delay duration to delay each {@link Subscriber#onNext} signal
+	 * @return a delayed {@link Flux}
+	 * @see #delaySubscription(Duration) delaySubscription to introduce a delay at the beginning of the sequence only
+	 */
+	public final Flux<T> delayElements(Duration delay) {
+		return delayElementsMillis(delay.toMillis());
+	}
+
+	/**
+	 * Delay each of this {@link Flux} elements ({@link Subscriber#onNext} signals)
+	 * by a given duration in milliseconds.
 	 *
-	 * @return a throttled {@link Flux}
+	 * <p>
+	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
 	 *
+	 * @param delay period to delay each {@link Subscriber#onNext} signal, in milliseconds
+	 * @deprecated will be replaced by {@link #delayElementsMillis(long)} in 3.1.0
+	 * @return a delayed {@link Flux}
 	 */
+	@Deprecated
 	public final Flux<T> delayMillis(long delay) {
-		return delayMillis(delay, Schedulers.timer());
+		return delayElementsMillis(delay);
 	}
 
 	/**
-	 * Delay this {@link Flux} signals to {@link Subscriber#onNext} until the given period in milliseconds elapses.
+	 * Delay each of this {@link Flux} elements ({@link Subscriber#onNext} signals)
+	 * by a given duration in milliseconds.
 	 *
 	 * <p>
 	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
 	 *
-	 * @param delay period to delay each {@link Subscriber#onNext} call in milliseconds
-	 * @param timer the timed scheduler to use for delaying each signal
+	 * @param delay period to delay each {@link Subscriber#onNext} signal, in milliseconds
+	 * @return a delayed {@link Flux}
+	 */
+	public final Flux<T> delayElementsMillis(long delay) {
+		return delayElementsMillis(delay, Schedulers.timer());
+	}
+
+	/**
+	 * Delay each of this {@link Flux} elements ({@link Subscriber#onNext} signals)
+	 * by a given duration in milliseconds.
 	 *
-	 * @return a throttled {@link Flux}
+	 * <p>
+	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
 	 *
+	 * @param delay period to delay each {@link Subscriber#onNext} signal, in milliseconds
+	 * @param timer the timed scheduler to use for delaying each signal
+	 * @deprecated will be replaced by {@link #delayElementsMillis(long, TimedScheduler)} in 3.1.0
+	 * @return a delayed {@link Flux}
 	 */
+	@Deprecated
 	public final Flux<T> delayMillis(long delay, TimedScheduler timer) {
+		return delayElementsMillis(delay, timer);
+	}
+
+	/**
+	 * Delay each of this {@link Flux} elements ({@link Subscriber#onNext} signals)
+	 * by a given duration in milliseconds.
+	 *
+	 * <p>
+	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
+	 *
+	 * @param delay period to delay each {@link Subscriber#onNext} signal, in milliseconds
+	 * @param timer the timed scheduler to use for delaying each signal
+	 * @return a delayed {@link Flux}
+	 */
+	public final Flux<T> delayElementsMillis(long delay, TimedScheduler timer) {
 		return concatMap(t ->  Mono.delayMillis(delay, timer).map(i -> t));
 	}
 

src/main/java/reactor/core/publisher/Mono.java

@@ -1328,6 +1328,63 @@ public final Mono<T> defaultIfEmpty(T defaultV) {
 		return onAssembly(new MonoDefaultIfEmpty<>(this, defaultV));
 	}
 
+	/**
+	 * Delay this {@link Flux} element ({@link Subscriber#onNext} signal) by a given
+	 * duration. Empty monos or error signals are not delayed.
+	 *
+	 * <p>
+	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
+	 *
+	 * <p>
+	 * Note that the scheduler on which the mono chain continues execution will be the
+	 * time scheduler used if the mono is valued, or the current scheduler if the mono
+	 * completes empty or errors.
+	 *
+	 * @param delay period to delay each {@link Subscriber#onNext} signal
+	 * @return a delayed {@link Mono}
+	 */
+	public final Mono<T> delayElement(Duration delay) {
+		return delayElementMillis(delay.toMillis());
+	}
+
+	/**
+	 * Delay this {@link Flux} element ({@link Subscriber#onNext} signal) by a given
+	 * duration, in milliseconds. Empty monos or error signals are not delayed.
+	 *
+	 * <p>
+	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
+	 *
+	 * <p>
+	 * Note that the scheduler on which the mono chain continues execution will be the
+	 * time scheduler used if the mono is valued, or the current scheduler if the mono
+	 * completes empty or errors.
+	 *
+	 * @param delay period to delay each {@link Subscriber#onNext} signal, in milliseconds
+	 * @return a delayed {@link Mono}
+	 */
+	public final Mono<T> delayElementMillis(long delay) {
+		return delayElementMillis(delay, Schedulers.timer());
+	}
+
+	/**
+	 * Delay this {@link Flux} element ({@link Subscriber#onNext} signal) by a given
+	 * duration, in milliseconds. Empty monos or error signals are not delayed.
+	 *
+	 * <p>
+	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/projectreactor.io/master/src/main/static/assets/img/marble/delayonnext.png" alt="">
+	 *
+	 * <p>
+	 * Note that the scheduler on which the mono chain continues execution will be the
+	 * time scheduler used if the mono is valued, or the current scheduler if the mono
+	 * completes empty or errors.
+	 *
+	 * @param delay period to delay each {@link Subscriber#onNext} signal, in milliseconds
+	 * @param timer the timed scheduler to use for delaying the value signal
+	 * @return a delayed {@link Mono}
+	 */
+	public final Mono<T> delayElementMillis(long delay, TimedScheduler timer) {
+		return onAssembly(new MonoDelayElement<>(this, delay, TimeUnit.MILLISECONDS, timer));
+	}
 
 	/**
 	 * Delay the {@link Mono#subscribe(Subscriber) subscription} to this {@link Mono} source until the given

src/main/java/reactor/core/publisher/MonoDelayElement.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright (c) 2011-2017 Pivotal Software Inc, 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.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor.core.publisher;
+
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import reactor.core.Cancellation;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.TimedScheduler;
+
+/**
+ * Emits the first value emitted by a given source publisher, delayed by some time amount
+ * with the help of a ScheduledExecutorService instance or a generic function callback that
+ * wraps other form of async-delayed execution of tasks.
+ *
+ * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
+ * @author Simon Baslé
+ */
+final class MonoDelayElement<T> extends MonoSource<T, T> {
+
+	final TimedScheduler timedScheduler;
+
+	final long delay;
+
+	final TimeUnit unit;
+
+	public MonoDelayElement(Publisher<? extends T> source, long delay, TimeUnit unit, TimedScheduler timedScheduler) {
+		super(source);
+		this.delay = delay;
+		this.unit = Objects.requireNonNull(unit, "unit");
+		this.timedScheduler = Objects.requireNonNull(timedScheduler, "timedScheduler");
+	}
+
+	@Override
+	public void subscribe(Subscriber<? super T> s) {
+		MonoDelayElementSubscriber r = new MonoDelayElementSubscriber<>(s, timedScheduler, delay, unit);
+		source.subscribe(r);
+	}
+
+	static final class MonoDelayElementSubscriber<T> extends Operators.MonoSubscriber<T,T>
+			implements Subscription  {
+
+		final long delay;
+		final TimedScheduler scheduler;
+		final TimeUnit unit;
+
+		Subscription s;
+
+		volatile Cancellation task;
+		volatile boolean done;
+
+		public MonoDelayElementSubscriber(Subscriber<? super T> actual, TimedScheduler scheduler,
+				long delay, TimeUnit unit) {
+			super(actual);
+			this.scheduler = scheduler;
+			this.delay = delay;
+			this.unit = unit;
+		}
+
+
+		@Override
+		public void cancel() {
+			super.cancel();
+			if (task != null) {
+				task.dispose();
+			}
+			if (s != Operators.cancelledSubscription()) {
+				s.cancel();
+			}
+		}
+
+		@Override
+		public void onSubscribe(Subscription s) {
+			if (Operators.validate(this.s, s)) {
+				this.s = s;
+
+				actual.onSubscribe(this);
+				s.request(1);
+			}
+		}
+
+		@Override
+		public void onNext(T t) {
+			if (done) {
+				Operators.onNextDropped(t);
+				return;
+			}
+			this.done = true;
+			this.task = scheduler.schedule(() -> complete(t), delay, unit);
+			if (task == Scheduler.REJECTED) {
+					throw Operators.onRejectedExecution(this, null, t);
+			}
+			else {
+				Subscription actualS = s;
+				s = Operators.cancelledSubscription();
+				actualS.cancel();
+			}
+		}
+
+		@Override
+		public void onComplete() {
+			if (done) {
+				return;
+			}
+			this.done = true;
+			actual.onComplete();
+		}
+
+		@Override
+		public void onError(Throwable t) {
+			if (done) {
+				Operators.onErrorDropped(t);
+				return;
+			}
+			this.done = true;
+			actual.onError(t);
+		}
+
+		@Override
+		public Object upstream() {
+			return s;
+		}
+	}
+}

src/test/java/reactor/core/publisher/FluxBufferBoundaryTest.java

@@ -246,7 +246,7 @@ public void bufferSupplierReturnsNUll() {
 
 	Flux<List<Integer>> scenario_bufferWillSubdivideAnInputFluxTime() {
 		return Flux.just(1, 2, 3, 4, 5, 6, 7, 8)
-		           .delay(Duration.ofMillis(99))
+		           .delayElements(Duration.ofMillis(99))
 		           .buffer(Duration.ofMillis(200));
 	}
 
@@ -263,7 +263,7 @@ public void bufferWillSubdivideAnInputFluxTime() {
 
 	Flux<List<Integer>> scenario_bufferWillSubdivideAnInputFluxTime2() {
 		return Flux.just(1, 2, 3, 4, 5, 6, 7, 8)
-		           .delay(Duration.ofMillis(99))
+		           .delayElements(Duration.ofMillis(99))
 		           .bufferMillis(200);
 	}
 

src/test/java/reactor/core/publisher/FluxBufferStartEndTest.java

@@ -162,7 +162,7 @@ public void bufferWillAcumulateMultipleListsOfValuesOverlap() {
 
 	Flux<List<Integer>> scenario_bufferWillSubdivideAnInputFluxOverlapTime() {
 		return Flux.just(1, 2, 3, 4, 5, 6, 7, 8)
-		           .delay(Duration.ofMillis(99))
+		           .delayElements(Duration.ofMillis(99))
 		           .buffer(Duration.ofMillis(300), Duration.ofMillis(200));
 	}
 
@@ -179,7 +179,7 @@ public void bufferWillSubdivideAnInputFluxOverlapTime() {
 
 	Flux<List<Integer>> scenario_bufferWillSubdivideAnInputFluxOverlapTime2() {
 		return Flux.just(1, 2, 3, 4, 5, 6, 7, 8)
-		           .delay(Duration.ofMillis(99))
+		           .delayElements(Duration.ofMillis(99))
 		           .bufferMillis(300L, 200L);//FIXME review signature
 	}
 
@@ -196,7 +196,7 @@ public void bufferWillSubdivideAnInputFluxOverlapTime2() {
 
 	Flux<List<Integer>> scenario_bufferWillSubdivideAnInputFluxSameTime() {
 		return Flux.just(1, 2, 3, 4, 5, 6, 7, 8)
-		           .delay(Duration.ofMillis(99))
+		           .delayElements(Duration.ofMillis(99))
 		           .bufferMillis(300L, 300L);
 	}
 
@@ -212,7 +212,7 @@ public void bufferWillSubdivideAnInputFluxSameTime() {
 
 	Flux<List<Integer>> scenario_bufferWillSubdivideAnInputFluxGapTime() {
 		return Flux.just(1, 2, 3, 4, 5, 6, 7, 8)
-		           .delay(Duration.ofMillis(99))
+		           .delayElements(Duration.ofMillis(99))
 		           .buffer(Duration.ofMillis(200), Duration.ofMillis(300));
 	}
 

src/test/java/reactor/core/publisher/FluxBufferTimeOrSizeTest.java

@@ -27,7 +27,7 @@ public class FluxBufferTimeOrSizeTest {
 
 	Flux<List<Integer>> scenario_bufferWithTimeoutAccumulateOnTimeOrSize() {
 		return Flux.range(1, 6)
-		           .delay(Duration.ofMillis(300))
+		           .delayElements(Duration.ofMillis(300))
 		           .buffer(5, Duration.ofMillis(2000));
 	}
 
@@ -43,7 +43,7 @@ public void bufferWithTimeoutAccumulateOnTimeOrSize() {
 
 	Flux<List<Integer>> scenario_bufferWithTimeoutAccumulateOnTimeOrSize2() {
 		return Flux.range(1, 6)
-		           .delay(Duration.ofMillis(300))
+		           .delayElements(Duration.ofMillis(300))
 		           .bufferMillis(5, 2000);
 	}