diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala index ca5dc1f208..bb1807626b 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala @@ -90,7 +90,7 @@ object FastFuture { def ready(atMost: Duration)(implicit permit: CanAwait) = this def transform[S](f: scala.util.Try[A] => scala.util.Try[S])( implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] = - FastFuture(f(Success(a))) + strictTransform(Success(a), f) def transformWith[S](f: scala.util.Try[A] => scala.concurrent.Future[S])( implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] = new FastFuture(this).transformWith(f) @@ -103,12 +103,20 @@ object FastFuture { def ready(atMost: Duration)(implicit permit: CanAwait) = this def transform[S](f: scala.util.Try[Nothing] => scala.util.Try[S])( implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] = - FastFuture(f(Failure(error))) + strictTransform(Failure(error), f) def transformWith[S](f: scala.util.Try[Nothing] => scala.concurrent.Future[S])( implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] = new FastFuture(this).transformWith(f) } + /** + * Applies `f` to an already available result, turning an exception thrown by `f` into a failed future + * the way [[scala.concurrent.Future.transform]] does for asynchronously completed futures. + */ + private def strictTransform[A, S](value: Try[A], f: Try[A] => Try[S]): Future[S] = + try FastFuture(f(value)) + catch { case NonFatal(e) => ErrorFuture(e) } + implicit class EnhancedFuture[T](val future: Future[T]) extends AnyVal { def fast: FastFuture[T] = new FastFuture[T](future) } diff --git a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala index b8b78b4c1b..fb3e0e474f 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala +++ b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala @@ -171,6 +171,37 @@ class FastFutureSpec extends AnyFreeSpec with Matchers { } } + "FastFuture-produced futures should fail instead of throwing when a standard Future combinator's function throws" - { + // The already completed futures override `Future.transform`, which the standard `filter`, `collect` and + // `transform` combinators are built on, so an exception thrown by the user function must become a + // failed future exactly as it does for a `Future.successful` / `Future.failed` value. + "transform on a successful future" in { + testStdlib(Success(23), _.transform(_ => throw TheException)) + } + "transform on a failed future" in { + testStdlib(Failure(UnexpectedException), _.transform(_ => throw TheException)) + } + "filter with a throwing predicate" in { + testStdlib(Success(23), _.filter(_ => throw TheException)) + } + "collect with a throwing partial function" in { + testStdlib(Success(23), _.collect { case _ => throw TheException }) + } + "map with a throwing function" in { + testStdlib(Success(23), _.map(_ => throw TheException)) + } + "recover with a throwing partial function" in { + testStdlib(Failure(UnexpectedException), _.recover { case _ => throw TheException }) + } + } + + def testStdlib(result: Try[Int], op: Future[Int] => Future[Int]): Unit = { + val f = FastFuture(result) + val transformed = op(f) + Await.ready(transformed, 500.millis) + transformed.value shouldEqual Some(Failure(TheException)) + } + def test(result: Try[Int], op: FastFuture[Int] => Future[Int])(check: Try[Int] => Unit): Unit = { def testStrictly(): Unit = { val f = FastFuture(result)