Merge pull request #11 from ruslansennov/v100

use snapshot vavr version

Author: ruslansennov

pom.xml

@@ -40,7 +40,7 @@
         <assertj.core.version>3.5.2</assertj.core.version>
         <eclipse.lifecycle.mapping.version>1.0.0</eclipse.lifecycle.mapping.version>
         <java.version>1.8</java.version>
-        <vavr.version>0.9.2</vavr.version>
+        <vavr.version>1.0.0-SNAPSHOT</vavr.version>
         <junit.version>4.12</junit.version>
         <gwt.version>2.8.0</gwt.version>
         <maven.bundle.version>3.2.0</maven.bundle.version>

src/main/resources/super/io/vavr/concurrent/CurrentThreadExecutorService.java

@@ -1,8 +1,21 @@
-/*                        __    __  __  __    __  ___
- *                       \  \  /  /    \  \  /  /  __/
- *                        \  \/  /  /\  \  \/  /  /
- *                         \____/__/  \__\____/__/.ɪᴏ
- * ᶜᵒᵖʸʳᶦᵍʰᵗ ᵇʸ ᵛᵃᵛʳ ⁻ ˡᶦᶜᵉⁿˢᵉᵈ ᵘⁿᵈᵉʳ ᵗʰᵉ ᵃᵖᵃᶜʰᵉ ˡᶦᶜᵉⁿˢᵉ ᵛᵉʳˢᶦᵒⁿ ᵗʷᵒ ᵈᵒᵗ ᶻᵉʳᵒ
+/*  __    __  __  __    __  ___
+ * \  \  /  /    \  \  /  /  __/
+ *  \  \/  /  /\  \  \/  /  /
+ *   \____/__/  \__\____/__/
+ *
+ * Copyright 2014-2017 Vavr, http://vavr.io
+ *
+ * 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.vavr.concurrent;
 

src/main/resources/super/io/vavr/concurrent/CurrentThreadFuture.java

@@ -1,8 +1,21 @@
-/*                        __    __  __  __    __  ___
- *                       \  \  /  /    \  \  /  /  __/
- *                        \  \/  /  /\  \  \/  /  /
- *                         \____/__/  \__\____/__/.ɪᴏ
- * ᶜᵒᵖʸʳᶦᵍʰᵗ ᵇʸ ᵛᵃᵛʳ ⁻ ˡᶦᶜᵉⁿˢᵉᵈ ᵘⁿᵈᵉʳ ᵗʰᵉ ᵃᵖᵃᶜʰᵉ ˡᶦᶜᵉⁿˢᵉ ᵛᵉʳˢᶦᵒⁿ ᵗʷᵒ ᵈᵒᵗ ᶻᵉʳᵒ
+/*  __    __  __  __    __  ___
+ * \  \  /  /    \  \  /  /  __/
+ *  \  \/  /  /\  \  \/  /  /
+ *   \____/__/  \__\____/__/
+ *
+ * Copyright 2014-2017 Vavr, http://vavr.io
+ *
+ * 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.vavr.concurrent;
 

src/main/resources/super/io/vavr/concurrent/FutureImpl.java

@@ -1,20 +1,34 @@
-/*                        __    __  __  __    __  ___
- *                       \  \  /  /    \  \  /  /  __/
- *                        \  \/  /  /\  \  \/  /  /
- *                         \____/__/  \__\____/__/.ɪᴏ
- * ᶜᵒᵖʸʳᶦᵍʰᵗ ᵇʸ ᵛᵃᵛʳ ⁻ ˡᶦᶜᵉⁿˢᵉᵈ ᵘⁿᵈᵉʳ ᵗʰᵉ ᵃᵖᵃᶜʰᵉ ˡᶦᶜᵉⁿˢᵉ ᵛᵉʳˢᶦᵒⁿ ᵗʷᵒ ᵈᵒᵗ ᶻᵉʳᵒ
+/*  __    __  __  __    __  ___
+ * \  \  /  /    \  \  /  /  __/
+ *  \  \/  /  /\  \  \/  /  /
+ *   \____/__/  \__\____/__/
+ *
+ * Copyright 2014-2017 Vavr, http://vavr.io
+ *
+ * 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.vavr.concurrent;
 
-import io.vavr.CheckedFunction0;
+import io.vavr.CheckedConsumer;
+import io.vavr.CheckedFunction1;
 import io.vavr.collection.Queue;
 import io.vavr.control.Option;
 import io.vavr.control.Try;
 
 import java.util.Objects;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.*;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 
 /**
  * GWT emulated version of {@link FutureImpl} with removed uses of Object's wait and notify methods.
@@ -31,18 +45,24 @@ final class FutureImpl<T> implements Future<T> {
      */
     private final Object lock = new Object();
 
+    /**
+     * Indicates if this Future is cancelled
+     */
+    @GuardedBy("lock")
+    private volatile boolean cancelled;
+
     /**
      * Once the Future is completed, the value is defined.
      */
     @GuardedBy("lock")
-    private volatile Option<Try<T>> value = Option.none();
+    private volatile Option<Try<T>> value;
 
     /**
      * The queue of actions is filled when calling onComplete() before the Future is completed or cancelled.
      * Otherwise actions = null.
      */
     @GuardedBy("lock")
-    private Queue<Consumer<? super Try<T>>> actions = Queue.empty();
+    private Queue<Consumer<Try<T>>> actions;
 
     /**
      * Once a computation is started via run(), job is defined and used to control the lifecycle of the computation.
@@ -51,16 +71,68 @@ final class FutureImpl<T> implements Future<T> {
      * {@code value} instead.
      */
     @GuardedBy("lock")
-    private java.util.concurrent.Future<?> job = null;
+    private java.util.concurrent.Future<?> job;
+
+    // single constructor
+    private FutureImpl(ExecutorService executorService, Option<Try<T>> value, Queue<Consumer<Try<T>>> actions, CheckedFunction1<FutureImpl<T>, java.util.concurrent.Future<?>> jobFactory) {
+        this.executorService = executorService;
+        synchronized (lock) {
+            this.cancelled = false;
+            this.value = value;
+            this.actions = actions;
+            try {
+                this.job = jobFactory.apply(this);
+            } catch(Throwable x) {
+                tryComplete(Try.failure(x));
+            }
+        }
+    }
 
     /**
-     * Creates a Future, {@link #run(CheckedFunction0)} has to be called separately.
+     * Creates a {@code FutureImpl} that is immediately completed with the given value. No task will be started.
      *
      * @param executorService An {@link ExecutorService} to run and control the computation and to perform the actions.
+     * @param value the result of this Future
      */
-    FutureImpl(ExecutorService executorService) {
-        Objects.requireNonNull(executorService, "executorService is null");
-        this.executorService = executorService;
+    @SuppressWarnings("unchecked")
+    static <T> FutureImpl<T> of(ExecutorService executorService, Try<? extends T> value) {
+        return new FutureImpl<>(executorService, Option.some(Try.narrow(value)), null, ignored -> null);
+    }
+
+    /**
+     * Creates a {@code FutureImpl} that is eventually completed.
+     * The given {@code computation} is <em>synchronously</em> executed, no thread is started.
+     *
+     * @param executorService  An {@link ExecutorService} to run and control the computation and to perform the actions.
+     * @param computation A non-blocking computation
+     * @param <T> value type of the Future
+     * @return a new {@code FutureImpl} instance
+     */
+    static <T> FutureImpl<T> sync(ExecutorService executorService, CheckedConsumer<Predicate<Try<? extends T>>> computation) {
+        return new FutureImpl<>(executorService, Option.none(), Queue.empty(), future -> {
+            computation.accept(future::tryComplete);
+            return null;
+        });
+    }
+
+    /**
+     * Creates a {@code FutureImpl} that is eventually completed.
+     * The given {@code computation} is <em>asynchronously</em> executed, a new thread is started.
+     *
+     * @param executorService  An {@link ExecutorService} to run and control the computation and to perform the actions.
+     * @param computation A (possibly blocking) computation
+     * @param <T> value type of the Future
+     * @return a new {@code FutureImpl} instance
+     */
+    static <T> FutureImpl<T> async(ExecutorService executorService, CheckedConsumer<Predicate<Try<? extends T>>> computation) {
+        // In a single-threaded context this Future may already have been completed during initialization.
+        return new FutureImpl<>(executorService, Option.none(), Queue.empty(), future -> executorService.submit(() -> {
+            try {
+                computation.accept(future::tryComplete);
+            } catch (Throwable x) {
+                future.tryComplete(Try.failure(x));
+            }
+        }));
     }
 
     @Override
@@ -69,20 +141,24 @@ public Future<T> await() {
     }
 
     @Override
-    public boolean cancel(boolean mayInterruptIfRunning) {
-        synchronized (lock) {
-            if (isCompleted()) {
-                return false;
-            } else {
-                return Try.of(() -> job == null || job.cancel(mayInterruptIfRunning)).onSuccess(cancelled -> {
-                    if (cancelled) {
-                        value = Option.some(Try.failure(new CancellationException()));
-                        actions = null;
-                        job = null;
-                    }
-                }).getOrElse(false);
+    public Future<T> await(long timeout, TimeUnit unit) {
+        return this;
+    }
+
+    @Override
+    public Future<T> cancel(boolean mayInterruptIfRunning) {
+        if (!isCompleted()) {
+            synchronized (lock) {
+                Try.of(() -> job == null || job.cancel(mayInterruptIfRunning))
+                        .recover(ignored -> job != null && job.isCancelled())
+                        .onSuccess(cancelled -> {
+                            if (cancelled) {
+                                this.cancelled = tryComplete(Try.failure(new CancellationException()));
+                            }
+                        });
             }
         }
+        return this;
     }
 
     @Override
@@ -95,11 +171,17 @@ public Option<Try<T>> getValue() {
         return value;
     }
 
+    @Override
+    public boolean isCancelled() {
+        return cancelled;
+    }
+
     @Override
     public boolean isCompleted() {
         return value.isDefined();
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public Future<T> onComplete(Consumer<? super Try<T>> action) {
         Objects.requireNonNull(action, "action is null");
@@ -110,7 +192,7 @@ public Future<T> onComplete(Consumer<? super Try<T>> action) {
                 if (isCompleted()) {
                     perform(action);
                 } else {
-                    actions = actions.enqueue(action);
+                    actions = actions.enqueue((Consumer<Try<T>>) action);
                 }
             }
         }
@@ -122,82 +204,54 @@ public Future<T> onComplete(Consumer<? super Try<T>> action) {
 
     @Override
     public String toString() {
-        return stringPrefix() + "(" + value.map(String::valueOf).getOrElse("?") + ")";
+        final Option<Try<T>> value = this.value;
+        final String s = (value == null || value.isEmpty()) ? "?" : value.get().toString();
+        return stringPrefix() + "(" + s + ")";
     }
 
     /**
-     * Runs a computation using the underlying ExecutorService.
+     * INTERNAL METHOD, SHOULD BE USED BY THE CONSTRUCTOR, ONLY.
      * <p>
-     * DEV-NOTE: Internally this method is called by the static {@code Future} factory methods.
-     *
-     * @throws IllegalStateException if the Future is pending, completed or cancelled
-     * @throws NullPointerException  if {@code computation} is null.
-     */
-    void run(CheckedFunction0<? extends T> computation) {
-        Objects.requireNonNull(computation, "computation is null");
-        synchronized (lock) {
-            if (job != null) {
-                throw new IllegalStateException("The Future is already running.");
-            }
-            if (isCompleted()) {
-                throw new IllegalStateException("The Future is completed.");
-            }
-            try {
-                // if the ExecutorService runs the computation
-                // - in a different thread, the lock ensures that the job is assigned before the computation completes
-                // - in the current thread, the job is already completed and the `job` variable remains null
-                final java.util.concurrent.Future<?> tmpJob = executorService.submit(() -> complete(Try.of(computation)));
-                if (!isCompleted()) {
-                    job = tmpJob;
-                }
-            } catch (Throwable t) {
-                // ensures that the Future completes if the `executorService.submit()` method throws
-                if (!isCompleted()) {
-                    complete(Try.failure(t));
-                }
-            }
-        }
-    }
-
-    boolean tryComplete(Try<? extends T> value) {
-        Objects.requireNonNull(value, "value is null");
-        synchronized (lock) {
-            if (isCompleted()) {
-                return false;
-            } else {
-                complete(value);
-                return true;
-            }
-        }
-    }
-
-    /**
-     * Completes this Future with a value.
+     * Completes this Future with a value and performs all actions.
      * <p>
-     * DEV-NOTE: Internally this method is called by the {@code Future.run()} method and by {@code Promise}.
+     * This method is idempotent. I.e. it does nothing, if this Future is already completed.
      *
      * @param value A Success containing a result or a Failure containing an Exception.
-     * @return The given {@code value} for convenience purpose.
      * @throws IllegalStateException if the Future is already completed or cancelled.
      * @throws NullPointerException  if the given {@code value} is null.
      */
-    private void complete(Try<? extends T> value) {
+    private boolean tryComplete(Try<? extends T> value) {
         Objects.requireNonNull(value, "value is null");
-        final Queue<Consumer<? super Try<T>>> actions;
-        // it is essential to make the completed state public *before* performing the actions
-        synchronized (lock) {
-            if (isCompleted()) {
-                throw new IllegalStateException("The Future is completed.");
+        if (isCompleted()) {
+            return false;
+        } else {
+            final Queue<Consumer<Try<T>>> actions;
+            // it is essential to make the completed state public *before* performing the actions
+            synchronized (lock) {
+                if (isCompleted()) {
+                    actions = null;
+                } else {
+                    // the job isn't set to null, see isCancelled()
+                    actions = this.actions;
+                    this.value = Option.some(Try.narrow(value));
+                    this.actions = null;
+                    this.job = null;
+                }
+            }
+            if (actions != null) {
+                actions.forEach(this::perform);
+                return true;
+            } else {
+                return false;
             }
-            actions = this.actions;
-            this.value = Option.some(Try.narrow(value));
-            this.actions = null;
-            this.job = null;
         }
-        actions.forEach(this::perform);
     }
 
     private void perform(Consumer<? super Try<T>> action) {
-        Try.run(() -> executorService.execute(() -> action.accept(value.get())));
+        try {
+            executorService.execute(() -> action.accept(value.get()));
+        } catch(Throwable x) {
+            // ignored // TODO: tell UncaughtExceptionHandler?
+        }
     }
 }