diff --git a/src/main/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapter.scala b/src/main/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapter.scala index 58069af8..2ef12bfc 100644 --- a/src/main/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapter.scala +++ b/src/main/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapter.scala @@ -15,10 +15,12 @@ */ 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 @@ -26,47 +28,15 @@ import scala.language.postfixOps * * @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) - } } diff --git a/src/test/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapterTest.scala b/src/test/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapterTest.scala index a6a21f7f..87423569 100644 --- a/src/test/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapterTest.scala +++ b/src/test/scala/com/comcast/xfinity/sirius/api/impl/compat/AkkaFutureAdapterTest.scala @@ -17,12 +17,17 @@ 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 { @@ -30,38 +35,68 @@ class AkkaFutureAdapterTest extends NiceTest with BeforeAndAfterAll { 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 + } + } + } }