Adapt Akka Future to Java 8 CompletableFuture

src/main/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapter.scala

@@ -15,58 +15,28 @@
  */
 package com.comcast.xfinity.sirius.api.impl.compat
 
-import scala.concurrent.{Await, Future => AkkaFuture}
-import scala.concurrent.duration._
-import java.util.concurrent.{TimeoutException, ExecutionException, TimeUnit, Future}
+import java.util.concurrent.CompletableFuture
+
+import scala.concurrent.Future
 import scala.language.postfixOps
+import scala.util.{Failure, Success}
+import scala.concurrent.ExecutionContext.Implicits.global
 
 /**
  * Class encapsulating a {@link akka.dispatch.Future} in a
  * {@link java.util.concurrent.Future}
  *
  * @param akkaFuture the {@see akka.dispatch.Future} to wrap
  */
-class AkkaFutureAdapter[T](akkaFuture: AkkaFuture[T]) extends Future[T] {
+class AkkaFutureAdapter[T](akkaFuture: Future[T]) extends CompletableFuture[T] {
+  akkaFuture onComplete {
+    case Success(result) => complete(result)
+    case Failure(exception) => completeExceptionally(exception)
+  }
 
   /**
-   * Not implemented, you may not cancel an Akka Future
-   */
-  def cancel(mayInterrupt: Boolean): Boolean =
+    * Not implemented, you may not cancel an Akka Future
+    */
+  override def cancel(mayInterruptIfRunning: Boolean): Boolean =
     throw new IllegalStateException("Not implemented")
-
-  /**
-   * Always returns true, since this is not cancellable
-   *
-   * @return true always
-   */
-  def isCancelled: Boolean = false // if not cancellable can't be cancelled
-
-  /**
-   * {@inheritDoc}
-   */
-  def isDone: Boolean = akkaFuture.isCompleted
-
-  /**
-   * {@inheritDoc}
-   */
-  def get: T =
-    try {
-      // there is still no way in hell this can time out
-      Await.result(akkaFuture, 7 days)
-    } catch {
-      case e: Throwable => throw new ExecutionException(e)
-    }
-
-  /**
-   * {@inheritDoc}
-   */
-  @throws(classOf[ExecutionException])
-  @throws(classOf[TimeoutException])
-  def get(l: Long, timeUnit: TimeUnit) =
-    try {
-      Await.result(akkaFuture, timeUnit.toMillis(l) milliseconds)
-    } catch {
-      case te: TimeoutException => throw te
-      case e: Throwable => throw new ExecutionException(e)
-    }
 }

src/test/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapterTest.scala

@@ -17,51 +17,86 @@ package com.comcast.xfinity.sirius.api.impl.compat
 
 import com.comcast.xfinity.sirius.NiceTest
 import org.scalatest.BeforeAndAfterAll
-import scala.concurrent.{ExecutionContext, Future => AkkaFuture}
-import akka.dispatch.ExecutionContexts._
+
+import scala.concurrent.{Await, Future}
 import akka.actor.ActorSystem
 import java.io.IOException
 import java.util.concurrent.{ExecutionException, TimeUnit, TimeoutException}
+import java.util.function.Consumer
+
 import akka.dispatch.ExecutionContexts
+import org.mockito.{Matchers, Mockito}
+
+import scala.concurrent.duration.Duration
 
 class AkkaFutureAdapterTest extends NiceTest with BeforeAndAfterAll {
 
   implicit val as = ActorSystem("AkkaFutureAdapterTest")
   implicit val ec = ExecutionContexts.global()
 
   override def afterAll {
-    as.shutdown()
-    as.awaitTermination()
+    Await.result(as.terminate(), Duration.Inf)
   }
 
   describe("AkkaFutureAdapter") {
     it("must throw an IllegalStateException when cancel is called") {
       intercept[IllegalStateException] {
-        val akkaFuture = AkkaFuture { "foo" }
+        val akkaFuture = Future { "foo" }
         new AkkaFutureAdapter[String](akkaFuture).cancel(true)
       }
     }
 
     it("must return the value expected on get") {
       assertResult("foo") {
-        val akkaFuture = AkkaFuture { "foo" }
+        val akkaFuture = Future { "foo" }
         new AkkaFutureAdapter[String](akkaFuture).get(2, TimeUnit.SECONDS)
       }
     }
 
     it("must throw a TimeoutException if it takes too long") {
       intercept[TimeoutException] {
-        val akkaFuture = AkkaFuture { Thread.sleep(1000); "foo" }
+        val akkaFuture = Future { Thread.sleep(1000); "foo" }
         new AkkaFutureAdapter[String](akkaFuture).get(500, TimeUnit.MILLISECONDS)
       }
     }
 
     it("must propogate an exception as an ExecutionException") {
       intercept[ExecutionException] {
-        val akkaFuture = AkkaFuture { throw new IOException("Boom") }
+        val akkaFuture = Future { throw new IOException("Boom") }
         new AkkaFutureAdapter[String](akkaFuture).get()
       }
     }
-  }
 
+    it("must complete the future with value") {
+      val akkaFuture = Future { "foo" }
+      val consumer = mock[Consumer[String]]
+      new AkkaFutureAdapter[String](akkaFuture).thenAccept(consumer)
+      Mockito.verify(consumer, Mockito.timeout(100).times(1))
+        .accept("foo")
+    }
+
+    it("must exceptionally complete the future with exception") {
+      assertResult("foo") {
+        val akkaFuture = Future {
+          throw new IOException("Boom")
+        }
+        val consumer = mock[Consumer[String]]
+        val function = mock[java.util.function.Function[Throwable, String]]
+        Mockito.doReturn("foo")
+          .when(function)
+          .apply(Matchers.any[Throwable])
+
+        val adapter = new AkkaFutureAdapter[String](akkaFuture)
+        adapter.thenAccept(consumer)
+        val continued = adapter.exceptionally(function)
+
+        Mockito.verify(function, Mockito.timeout(100).times(1))
+          .apply(Matchers.any[IOException])
+        Mockito.verify(consumer, Mockito.never())
+          .accept(Matchers.anyString)
+
+        continued.get
+      }
+    }
+  }
 }