[Reactive] Implement timer() + TCK tests (helidon-io#1516)

* [Reactive] Implement timer() + TCK tests

* Add missing copyright

Author: akarnokd

common/reactive/src/main/java/io/helidon/common/reactive/Multi.java

@@ -22,6 +22,8 @@
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Publisher;
 import java.util.concurrent.Flow.Subscriber;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -515,6 +517,20 @@ static Multi<Long> rangeLong(long start, long count) {
         return new MultiRangeLongPublisher(start, start + count);
     }
 
+    /**
+     * Signal 0L and complete the sequence after the given time elapsed.
+     * @param time the time to wait before signaling 0L and completion
+     * @param unit the unit of time
+     * @param executor the executor to run the waiting on
+     * @return Multi
+     * @throws NullPointerException if {@code unit} or {@code executor} is {@code null}
+     */
+    static Multi<Long> timer(long time, TimeUnit unit, ScheduledExecutorService executor) {
+        Objects.requireNonNull(unit, "unit is null");
+        Objects.requireNonNull(executor, "executor is null");
+        return new MultiTimer(time, unit, executor);
+    }
+
     /**
      * {@link java.util.function.Function} providing one item to be submitted as onNext in case of onError signal is received.
      *

common/reactive/src/main/java/io/helidon/common/reactive/MultiTimer.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright (c)  2020 Oracle and/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.
+ * 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 io.helidon.common.reactive;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Flow;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Signal 0L and complete after the specified time.
+ */
+final class MultiTimer implements Multi<Long> {
+
+    private final long time;
+
+    private final TimeUnit unit;
+
+    private final ScheduledExecutorService executor;
+
+    MultiTimer(long time, TimeUnit unit, ScheduledExecutorService executor) {
+        this.time = time;
+        this.unit = unit;
+        this.executor = executor;
+    }
+
+    @Override
+    public void subscribe(Flow.Subscriber<? super Long> subscriber) {
+        TimerSubscription subscription = new TimerSubscription(subscriber);
+        subscriber.onSubscribe(subscription);
+
+        subscription.setFuture(executor.schedule(subscription, time, unit));
+    }
+
+    static final class TimerSubscription extends DeferredScalarSubscription<Long>
+    implements Callable<Void> {
+
+        private final AtomicReference<Future<?>> future;
+
+        TimerSubscription(Flow.Subscriber<? super Long> downstream) {
+            super(downstream);
+            this.future = new AtomicReference<>();
+        }
+
+        @Override
+        public Void call() {
+            future.lazySet(TerminatedFuture.FINISHED);
+            complete(0L);
+            return null;
+        }
+
+        @Override
+        public void cancel() {
+            super.cancel();
+            TerminatedFuture.cancel(future);
+        }
+
+        public void setFuture(Future<?> f) {
+            TerminatedFuture.setFuture(future, f);
+        }
+    }
+
+}

common/reactive/src/main/java/io/helidon/common/reactive/Single.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Publisher;
 import java.util.concurrent.Flow.Subscriber;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
@@ -230,6 +231,21 @@ static <T> Single<T> never() {
         return SingleNever.<T>instance();
     }
 
+
+    /**
+     * Signal 0L and complete the sequence after the given time elapsed.
+     * @param time the time to wait before signaling 0L and completion
+     * @param unit the unit of time
+     * @param executor the executor to run the waiting on
+     * @return Single
+     * @throws NullPointerException if {@code unit} or {@code executor} is {@code null}
+     */
+    static Single<Long> timer(long time, TimeUnit unit, ScheduledExecutorService executor) {
+        Objects.requireNonNull(unit, "unit is null");
+        Objects.requireNonNull(executor, "executor is null");
+        return new SingleTimer(time, unit, executor);
+    }
+
     /**
      * Relay upstream items until the other source signals an item or completes.
      * @param other the other sequence to signal the end of the main sequence

common/reactive/src/main/java/io/helidon/common/reactive/SingleTimer.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c)  2020 Oracle and/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.
+ * 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 io.helidon.common.reactive;
+
+import java.util.concurrent.Flow;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Signal 0L and complete after the specified time.
+ */
+final class SingleTimer implements Single<Long> {
+
+    private final long time;
+
+    private final TimeUnit unit;
+
+    private final ScheduledExecutorService executor;
+
+    SingleTimer(long time, TimeUnit unit, ScheduledExecutorService executor) {
+        this.time = time;
+        this.unit = unit;
+        this.executor = executor;
+    }
+
+    @Override
+    public void subscribe(Flow.Subscriber<? super Long> subscriber) {
+        MultiTimer.TimerSubscription subscription = new MultiTimer.TimerSubscription(subscriber);
+        subscriber.onSubscribe(subscription);
+
+        subscription.setFuture(executor.schedule(subscription, time, unit));
+    }
+}

common/reactive/src/main/java/io/helidon/common/reactive/TerminatedFuture.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright (c)  2020 Oracle and/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.
+ * 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 io.helidon.common.reactive;
+
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Enumeration that implements a finished or canceled future to be used
+ * for reference comparison and atomic state transitions.
+ */
+enum TerminatedFuture implements Future<Object> {
+
+    /** The task has already finished normally. */
+    FINISHED,
+
+    /** The task has been cancelled. */
+    CANCELED;
+
+    @Override
+    public boolean cancel(boolean mayInterruptIfRunning) {
+        return false;
+    }
+
+    @Override
+    public boolean isCancelled() {
+        return this == CANCELED;
+    }
+
+    @Override
+    public boolean isDone() {
+        return true;
+    }
+
+    @Override
+    public Object get() {
+        if (this == CANCELED) {
+            throw new CancellationException();
+        }
+        return null;
+    }
+
+    @Override
+    public Object get(long timeout, TimeUnit unit) {
+        if (this == CANCELED) {
+            throw new CancellationException();
+        }
+        return null;
+    }
+
+    /**
+     * Atomically set a {@link Future} on the given field if that field doesn't
+     * already hold the {@link #CANCELED} or {@link #FINISHED} instance.
+     * @param field the target atomic reference field
+     * @param f the future to set
+     */
+    public static void setFuture(AtomicReference<Future<?>> field, Future<?> f) {
+        for (;;) {
+            Future<?> current = field.get();
+            if (current == TerminatedFuture.CANCELED) {
+                f.cancel(true);
+                return;
+            }
+            if (current != null) {
+                return;
+            }
+            if (field.compareAndSet(null, f)) {
+                return;
+            }
+        }
+    }
+
+    /**
+     * Atomically cancel the future in the target field or mark it for
+     * cancellation.
+     * @param field the target atomic reference field
+     */
+    public static void cancel(AtomicReference<Future<?>> field) {
+        for (;;) {
+            Future<?> current = field.get();
+            if (current == TerminatedFuture.FINISHED || current == TerminatedFuture.CANCELED) {
+                return;
+            }
+            if (field.compareAndSet(current, TerminatedFuture.CANCELED)) {
+                if (current != null) {
+                    current.cancel(true);
+                }
+                return;
+            }
+        }
+    }
+}

common/reactive/src/test/java/io/helidon/common/reactive/MultiTimerTckTest.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2020 Oracle and/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.
+ * 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 io.helidon.common.reactive;
+
+import org.reactivestreams.tck.TestEnvironment;
+import org.reactivestreams.tck.flow.FlowPublisherVerification;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.Flow;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Test
+public class MultiTimerTckTest extends FlowPublisherVerification<Long> {
+
+    private static ScheduledExecutorService executor;
+
+    public MultiTimerTckTest() {
+        super(new TestEnvironment(200));
+    }
+
+    @BeforeClass
+    public static void beforeClass() {
+        executor = Executors.newSingleThreadScheduledExecutor();
+    }
+
+    @AfterClass
+    public static void afterClass() {
+        executor.shutdown();
+    }
+
+    @Override
+    public Flow.Publisher<Long> createFlowPublisher(long l) {
+        return Multi.timer(1, TimeUnit.MILLISECONDS, executor);
+    }
+
+    @Override
+    public Flow.Publisher<Long> createFailedFlowPublisher() {
+        return null;
+    }
+
+    @Override
+    public long maxElementsFromPublisher() {
+        return 1;
+    }
+}