Rework for thread safety

eclipse-vertx · Jan 15, 2016 · 374c599 · 374c599
1 parent 4a4b1d0
commit 374c599
Show file tree

Hide file tree

Showing 4 changed files with 163 additions and 44 deletions.
diff --git a/src/main/asciidoc/java/index.adoc b/src/main/asciidoc/java/index.adoc
@@ -331,6 +331,8 @@ CompositeFuture.any(future1, future2).setHandler(ar -> {
 });
 ----
 
+`link:../../apidocs/io/vertx/core/Future.html#compose-io.vertx.core.Handler-io.vertx.core.Future-[compose]` chains futures:
+
 == Verticles
 
 Vert.x comes with a simple, scalable, _actor-like_ deployment and concurrency model out of the box that

diff --git a/src/main/java/io/vertx/core/Future.java b/src/main/java/io/vertx/core/Future.java
@@ -147,7 +147,22 @@ static <T> Future<T> failedFuture(String failureMessage) {
    * @param handler the handler
    * @param next the next future
    */
-  <U> void compose(Handler<T> handler, Future<U> next);
+  default <U> void compose(Handler<T> handler, Future<U> next) {
+    setHandler(ar -> {
+      if (ar.succeeded()) {
+        try {
+          handler.handle(ar.result());
+        } catch (Throwable err) {
+          if (next.isComplete()) {
+            throw err;
+          }
+          next.fail(err);
+        }
+      } else {
+        next.fail(ar.cause());
+      }
+    });
+  }
 
   /**
    * @return an handler completing this future

diff --git a/src/main/java/io/vertx/core/impl/CompositeFutureImpl.java b/src/main/java/io/vertx/core/impl/CompositeFutureImpl.java
@@ -24,23 +24,31 @@
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public class CompositeFutureImpl extends FutureImpl<CompositeFuture> implements CompositeFuture {
+public class CompositeFutureImpl implements CompositeFuture, Handler<AsyncResult<CompositeFuture>> {
 
   public static CompositeFuture all(Future<?>... results) {
     CompositeFutureImpl composite = new CompositeFutureImpl(results);
     for (int i = 0;i < results.length;i++) {
       int index = i;
       results[i].setHandler(ar -> {
+        Handler<AsyncResult<CompositeFuture>> handler = null;
         if (ar.succeeded()) {
-          composite.flag |= 1 << index;
+          synchronized (composite) {
-          if (!composite.isComplete() && composite.flag == (1 << results.length) - 1) {
+            composite.flag |= 1 << index;
-            composite.complete(composite);
+            if (!composite.isComplete() && composite.flag == (1 << results.length) - 1) {
+              handler = composite.setSucceeded();
+            }
           }
         } else {
-          if (!composite.isComplete()) {
+          synchronized (composite) {
-            composite.fail(ar.cause());
+            if (!composite.isComplete()) {
+              handler = composite.setFailed(ar.cause());
+            }
           }
         }
+        if (handler != null) {
+          handler.handle(composite);
+        }
       });
     }
     return composite;
@@ -51,63 +59,172 @@ public static CompositeFuture any(Future<?>... results) {
     for (int i = 0;i < results.length;i++) {
       int index = i;
       results[i].setHandler(ar -> {
+        Handler<AsyncResult<CompositeFuture>> handler = null;
         if (ar.succeeded()) {
-          if (!composite.isComplete()) {
+          synchronized (composite) {
-            composite.complete(composite);
+            if (!composite.isComplete()) {
+              composite.setSucceeded();
+            }
           }
         } else {
-          composite.flag |= 1 << index;
+          synchronized (composite) {
-          if (!composite.isComplete() && composite.flag == (1 << results.length) - 1) {
+            composite.flag |= 1 << index;
-            composite.fail(ar.cause());
+            if (!composite.isComplete() && composite.flag == (1 << results.length) - 1) {
+              handler = composite.setFailed(ar.cause());
+            }
           }
         }
+        if (handler != null) {
+          handler.handle(composite);
+        }
       });
     }
     return composite;
   }
 
   private final Future[] results;
   private int flag;
+  private boolean completed;
+  private Throwable cause;
+  private Handler<AsyncResult<CompositeFuture>> handler;
 
   private CompositeFutureImpl(Future<?>... results) {
     this.results = results;
   }
 
   @Override
   public CompositeFuture setHandler(Handler<AsyncResult<CompositeFuture>> handler) {
-    return (CompositeFuture) super.setHandler(handler);
+    boolean call;
+    synchronized (this) {
+      this.handler = handler;
+      call = completed;
+    }
+    if (call) {
+      handler.handle(this);
+    }
+    return this;
   }
 
   @Override
   public Throwable cause(int index) {
-    return results[index].cause();
+    return future(index).cause();
   }
 
   @Override
   public boolean succeeded(int index) {
-    return results[index].succeeded();
+    return future(index).succeeded();
   }
 
   @Override
   public boolean failed(int index) {
-    return results[index].failed();
+    return future(index).failed();
   }
 
   @Override
   public boolean isComplete(int index) {
-    return results[index].isComplete();
+    return future(index).isComplete();
   }
 
   @Override
   public <T> T result(int index) {
+    return this.<T>future(index).result();
+  }
+
+  private <T> Future<T> future(int index) {
     if (index < 0 || index > results.length) {
       throw new IndexOutOfBoundsException();
     }
-    return (T) results[index].result();
+    return (Future<T>) results[index];
   }
 
   @Override
   public int size() {
     return results.length;
   }
+
+  @Override
+  public synchronized boolean isComplete() {
+    return completed;
+  }
+
+  @Override
+  public synchronized boolean succeeded() {
+    return completed && cause == null;
+  }
+
+  @Override
+  public synchronized boolean failed() {
+    return completed && cause != null;
+  }
+
+  @Override
+  public synchronized Throwable cause() {
+    return completed && cause != null ? cause : null;
+  }
+
+  @Override
+  public synchronized CompositeFuture result() {
+    return completed && cause == null ? this : null;
+  }
+
+  @Override
+  public void complete(CompositeFuture result) {
+    Handler<AsyncResult<CompositeFuture>> handler = setSucceeded();
+    if (handler != null) {
+      handler.handle(this);
+    }
+  }
+
+  @Override
+  public void complete() {
+    complete(null);
+  }
+
+  @Override
+  public void fail(Throwable throwable) {
+    Handler<AsyncResult<CompositeFuture>> handler = setFailed(throwable);
+    if (handler != null) {
+      handler.handle(this);
+    }
+  }
+
+  @Override
+  public void fail(String failureMessage) {
+    fail(new NoStackTraceThrowable(failureMessage));
+  }
+
+  private Handler<AsyncResult<CompositeFuture>> setFailed(Throwable cause) {
+    synchronized (this) {
+      if (completed) {
+        throw new IllegalStateException("Result is already complete: " + (this.cause == null ? "succeeded" : "failed"));
+      }
+      this.completed = true;
+      this.cause = cause;
+      return handler;
+    }
+  }
+
+  private Handler<AsyncResult<CompositeFuture>> setSucceeded() {
+    synchronized (this) {
+      if (completed) {
+        throw new IllegalStateException("Result is already complete: " + (this.cause == null ? "succeeded" : "failed"));
+      }
+      this.completed = true;
+      return handler;
+    }
+  }
+
+  @Override
+  public Handler<AsyncResult<CompositeFuture>> handler() {
+    return this;
+  }
+
+  @Override
+  public void handle(AsyncResult<CompositeFuture> ar) {
+    if (ar.succeeded()) {
+      complete(this);
+    } else {
+      fail(ar.cause());
+    }
+  }
 }
diff --git a/src/main/java/io/vertx/core/impl/FutureImpl.java b/src/main/java/io/vertx/core/impl/FutureImpl.java
@@ -20,7 +20,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 
-class FutureImpl<T> implements Future<T> {
+class FutureImpl<T> implements Future<T>, Handler<AsyncResult<T>> {
   private boolean failed;
   private boolean succeeded;
   private Handler<AsyncResult<T>> handler;
@@ -126,13 +126,16 @@ public void handle(Future<T> ar) {
 
   @Override
   public Handler<AsyncResult<T>> handler() {
-    return ar -> {
+    return this;
-      if (ar.succeeded()) {
+  }
-        complete(ar.result());
+
-      } else {
+  @Override
-        fail(ar.cause());
+  public void handle(AsyncResult<T> ar) {
-      }
+    if (ar.succeeded()) {
-    };
+      complete(ar.result());
+    } else {
+      fail(ar.cause());
+    }
   }
 
   /**
@@ -145,24 +148,6 @@ public void fail(Throwable throwable) {
     checkCallHandler();
   }
 
-  @Override
-  public <U> void compose(Handler<T> handler, Future<U> next) {
-    setHandler(ar -> {
-      if (ar.succeeded()) {
-        try {
-          handler.handle(ar.result());
-        } catch (Throwable err) {
-          if (next.isComplete()) {
-            throw err;
-          }
-          next.fail(err);
-        }
-      } else {
-        next.fail(ar.cause());
-      }
-    });
-  }
-
   @Override
   public void fail(String failureMessage) {
     fail(new NoStackTraceThrowable(failureMessage));