+act #20936 add CompletionStage API to CircuitBreaker (#20937)

akka-actor-tests/src/test/java/akka/pattern/CircuitBreakerTest.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright (C) 2009-2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.pattern;
+
+import akka.actor.*;
+import akka.testkit.AkkaJUnitActorSystemResource;
+import akka.testkit.AkkaSpec;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.scalatest.junit.JUnitSuite;
+import scala.compat.java8.FutureConverters;
+import scala.concurrent.Await;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class CircuitBreakerTest extends JUnitSuite {
+
+  @ClassRule
+  public static AkkaJUnitActorSystemResource actorSystemResource = 
+    new AkkaJUnitActorSystemResource("JavaAPI", AkkaSpec.testConf());
+
+  private final ActorSystem system = actorSystemResource.getSystem();
+
+  @Test
+  public void useCircuitBreakerWithCompletableFuture() throws Exception {
+    final FiniteDuration fiveSeconds = FiniteDuration.create(5, TimeUnit.SECONDS);
+    final FiniteDuration fiveHundredMillis = FiniteDuration.create(500, TimeUnit.MILLISECONDS);
+    final CircuitBreaker breaker = new CircuitBreaker(system.dispatcher(), system.scheduler(), 1, fiveSeconds, fiveHundredMillis);
+
+    final CompletableFuture<String> f = new CompletableFuture<>();
+    f.complete("hello");
+    final CompletionStage<String> res = breaker.callWithCircuitBreakerCS(() -> f);
+    assertEquals("hello", Await.result(FutureConverters.toScala(res), fiveSeconds));
+  }
+
+}

akka-actor-tests/src/test/scala/akka/pattern/CircuitBreakerSpec.scala

@@ -219,7 +219,7 @@ class CircuitBreakerSpec extends AkkaSpec with BeforeAndAfter {
       val breaker = CircuitBreakerSpec.shortCallTimeoutCb()
 
       val fut = breaker().withCircuitBreaker(Future {
-        Thread.sleep(150.millis.dilated.toMillis);
+        Thread.sleep(150.millis.dilated.toMillis)
         throwException
       })
       checkLatch(breaker.openLatch)

akka-actor/src/main/scala/akka/pattern/CircuitBreaker.scala

@@ -3,18 +3,24 @@
  */
 package akka.pattern
 
-import java.util.concurrent.atomic.{ AtomicInteger, AtomicLong, AtomicBoolean }
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger, AtomicLong }
+
 import akka.AkkaException
 import akka.actor.Scheduler
 import akka.util.Unsafe
+
 import scala.util.control.NoStackTrace
-import java.util.concurrent.{ Callable, CopyOnWriteArrayList }
+import java.util.concurrent.{ Callable, CompletionStage, CopyOnWriteArrayList }
-import scala.concurrent.{ ExecutionContext, Future, Promise, Await }
+
+import scala.concurrent.{ Await, ExecutionContext, Future, Promise }
 import scala.concurrent.duration._
 import scala.concurrent.TimeoutException
 import scala.util.control.NonFatal
 import scala.util.Success
 import akka.dispatch.ExecutionContexts.sameThreadExecutionContext
+import akka.japi.function.Creator
+
+import scala.compat.java8.FutureConverters
 
 /**
  * Companion object providing factory methods for Circuit Breaker which runs callbacks in caller's thread
@@ -123,6 +129,18 @@ class CircuitBreaker(scheduler: Scheduler, maxFailures: Int, callTimeout: Finite
    */
   def callWithCircuitBreaker[T](body: Callable[Future[T]]): Future[T] = withCircuitBreaker(body.call)
 
+  /**
+   * Java API (8) for [[#withCircuitBreaker]]
+   *
+   * @param body Call needing protected
+   * @return [[java.util.concurrent.CompletionStage]] containing the call result or a
+   *   `scala.concurrent.TimeoutException` if the call timed out
+   */
+  def callWithCircuitBreakerCS[T](body: Callable[CompletionStage[T]]): CompletionStage[T] =
+    FutureConverters.toJava[T](callWithCircuitBreaker(new Callable[Future[T]] {
+      override def call(): Future[T] = FutureConverters.toScala(body.call())
+    }))
+
   /**
    * Wraps invocations of synchronous calls that need to be protected
    *

akka-docs/rst/common/code/docs/circuitbreaker/DangerousJavaActor.java

@@ -51,27 +51,16 @@ public class DangerousJavaActor extends UntypedActor {
     if (message instanceof String) {
       String m = (String) message;
       if ("is my middle name".equals(m)) {
-        pipe(breaker.callWithCircuitBreaker(
+        pipe(
-          new Callable<Future<String>>() {
+          breaker.callWithCircuitBreaker(() -> 
-            public Future<String> call() throws Exception {
+            future(() -> dangerousCall(), getContext().dispatcher())
-              return future(
+          ), getContext().dispatcher()
-                      new Callable<String>() {
+        ).to(getSender());
-                          public String call() {
-                            return dangerousCall();
-                          }
-                        }, getContext().dispatcher());
-            }
-          }), getContext().dispatcher()).to(getSender());
       }
       if ("block for me".equals(m)) {
         getSender().tell(breaker
           .callWithSyncCircuitBreaker(
-            new Callable<String>() {
+            () -> dangerousCall()), getSelf());
-              @Override
-              public String call() throws Exception {
-                return dangerousCall();
-              }
-            }), getSelf());
       }
     }
   }