Allow to pass an Executor when converting Futures

This is to circumvent #179.

Author: rdesgroppes

src/main/scala/scala/compat/java8/FutureConverters.scala

@@ -17,7 +17,7 @@ import scala.language.implicitConversions
 import scala.concurrent.java8.FuturesConvertersImpl._
 import scala.concurrent.java8.FuturesConvertersImplCompat._
 import scala.concurrent.{ Future, Promise, ExecutionContext, ExecutionContextExecutorService, ExecutionContextExecutor }
-import java.util.concurrent.{ CompletionStage, Executor, ExecutorService }
+import java.util.concurrent.{ CompletionStage, Executor, ExecutorService, ForkJoinPool }
 import java.util.function.Consumer
 
 /**
@@ -59,16 +59,38 @@ object FutureConverters {
    * transformations to their asynchronous counterparts, i.e.
    * <code>thenRun</code> will internally call <code>thenRunAsync</code>.
    *
+   * Callbacks will run onto ForkJoinPool.commonPool(), unless it does not
+   * support a parallelism level of at least two, in which case a new Thread
+   * is used.
+   *
    * @param f The Scala Future which may eventually supply the completion for
    * the returned CompletionStage
    * @return a CompletionStage that runs all callbacks asynchronously and does
    * not support the CompletableFuture interface
    */
-  def toJava[T](f: Future[T]): CompletionStage[T] = {
+  def toJava[T](f: Future[T]): CompletionStage[T] = toJava(f, ForkJoinPool.commonPool())
+
+  /**
+    * Returns a CompletionStage that will be completed with the same value or
+    * exception as the given Scala Future when that completes. Since the Future is a read-only
+    * representation, this CompletionStage does not support the
+    * <code>toCompletableFuture</code> method. The semantics of Scala Future
+    * demand that all callbacks are invoked asynchronously by default, therefore
+    * the returned CompletionStage routes all calls to synchronous
+    * transformations to their asynchronous counterparts, i.e.
+    * <code>thenRun</code> will internally call <code>thenRunAsync</code>.
+    *
+    * @param f The Scala Future which may eventually supply the completion for
+    * the returned CompletionStage
+    * @param e The Java Executor onto which run the callbacks
+    * @return a CompletionStage that runs all callbacks asynchronously and does
+    * not support the CompletableFuture interface
+    */
+  def toJava[T](f: Future[T], e: Executor): CompletionStage[T] = {
     f match {
       case p: P[T @unchecked] => p.wrapped
       case _ =>
-        val cf = new CF[T](f)
+        val cf = new CF[T](f, e)
         implicit val ec = InternalCallbackExecutor
         f onComplete cf
         cf
@@ -189,10 +211,30 @@ object FutureConverters {
      * transformations to their asynchronous counterparts, i.e.
      * <code>thenRun</code> will internally call <code>thenRunAsync</code>.
      *
+     * Callbacks will run onto ForkJoinPool.commonPool(), unless it does not
+     * support a parallelism level of at least two, in which case a new Thread
+     * is used.
+     *
      * @return a CompletionStage that runs all callbacks asynchronously and does
      * not support the CompletableFuture interface
      */
     def toJava: CompletionStage[T] = FutureConverters.toJava(__self)
+
+    /**
+      * Returns a CompletionStage that will be completed with the same value or
+      * exception as the given Scala Future when that completes. Since the Future is a read-only
+      * representation, this CompletionStage does not support the
+      * <code>toCompletableFuture</code> method. The semantics of Scala Future
+      * demand that all callbacks are invoked asynchronously by default, therefore
+      * the returned CompletionStage routes all calls to synchronous
+      * transformations to their asynchronous counterparts, i.e.
+      * <code>thenRun</code> will internally call <code>thenRunAsync</code>.
+      *
+      * @param e The Java Executor onto which run the callbacks
+      * @return a CompletionStage that runs all callbacks asynchronously and does
+      * not support the CompletableFuture interface
+      */
+    def toJava(e: Executor): CompletionStage[T] = FutureConverters.toJava(__self, e)
   }
 
   implicit def CompletionStageOps[T](cs: CompletionStage[T]): CompletionStageOps[T] = new CompletionStageOps(cs)

src/main/scala/scala/concurrent/java8/FutureConvertersImpl.scala

@@ -23,7 +23,9 @@ import scala.util.{Failure, Success, Try}
 
 // TODO: make this private[scala] when genjavadoc allows for that.
 object FuturesConvertersImpl {
-  class CF[T](val wrapped: Future[T]) extends CompletableFuture[T] with (Try[T] => Unit) {
+  class CF[T](val wrapped: Future[T], executor: Executor) extends CompletableFuture[T] with (Try[T] => Unit) {
+    def this(wrapped: Future[T]) = this(wrapped, ForkJoinPool.commonPool())
+
     override def apply(t: Try[T]): Unit = t match {
       case Success(v) => complete(v)
       case Failure(e) => completeExceptionally(e)
@@ -32,29 +34,29 @@ object FuturesConvertersImpl {
     /*
      * Ensure that completions of this future cannot hold the Scala Future’s completer hostage.
      */
-    override def thenApply[U](fn: JF[_ >: T, _ <: U]): CompletableFuture[U] = thenApplyAsync(fn)
+    override def thenApply[U](fn: JF[_ >: T, _ <: U]): CompletableFuture[U] = thenApplyAsync(fn, executor)
 
-    override def thenAccept(fn: Consumer[_ >: T]): CompletableFuture[Void] = thenAcceptAsync(fn)
+    override def thenAccept(fn: Consumer[_ >: T]): CompletableFuture[Void] = thenAcceptAsync(fn, executor)
 
-    override def thenRun(fn: Runnable): CompletableFuture[Void] = thenRunAsync(fn)
+    override def thenRun(fn: Runnable): CompletableFuture[Void] = thenRunAsync(fn, executor)
 
-    override def thenCombine[U, V](cs: CompletionStage[_ <: U], fn: BiFunction[_ >: T, _ >: U, _ <: V]): CompletableFuture[V] = thenCombineAsync(cs, fn)
+    override def thenCombine[U, V](cs: CompletionStage[_ <: U], fn: BiFunction[_ >: T, _ >: U, _ <: V]): CompletableFuture[V] = thenCombineAsync(cs, fn, executor)
 
-    override def thenAcceptBoth[U](cs: CompletionStage[_ <: U], fn: BiConsumer[_ >: T, _ >: U]): CompletableFuture[Void] = thenAcceptBothAsync(cs, fn)
+    override def thenAcceptBoth[U](cs: CompletionStage[_ <: U], fn: BiConsumer[_ >: T, _ >: U]): CompletableFuture[Void] = thenAcceptBothAsync(cs, fn, executor)
 
-    override def runAfterBoth(cs: CompletionStage[_], fn: Runnable): CompletableFuture[Void] = runAfterBothAsync(cs, fn)
+    override def runAfterBoth(cs: CompletionStage[_], fn: Runnable): CompletableFuture[Void] = runAfterBothAsync(cs, fn, executor)
 
-    override def applyToEither[U](cs: CompletionStage[_ <: T], fn: JF[_ >: T, U]): CompletableFuture[U] = applyToEitherAsync(cs, fn)
+    override def applyToEither[U](cs: CompletionStage[_ <: T], fn: JF[_ >: T, U]): CompletableFuture[U] = applyToEitherAsync(cs, fn, executor)
 
-    override def acceptEither(cs: CompletionStage[_ <: T], fn: Consumer[_ >: T]): CompletableFuture[Void] = acceptEitherAsync(cs, fn)
+    override def acceptEither(cs: CompletionStage[_ <: T], fn: Consumer[_ >: T]): CompletableFuture[Void] = acceptEitherAsync(cs, fn, executor)
 
-    override def runAfterEither(cs: CompletionStage[_], fn: Runnable): CompletableFuture[Void] = runAfterEitherAsync(cs, fn)
+    override def runAfterEither(cs: CompletionStage[_], fn: Runnable): CompletableFuture[Void] = runAfterEitherAsync(cs, fn, executor)
 
-    override def thenCompose[U](fn: JF[_ >: T, _ <: CompletionStage[U]]): CompletableFuture[U] = thenComposeAsync(fn)
+    override def thenCompose[U](fn: JF[_ >: T, _ <: CompletionStage[U]]): CompletableFuture[U] = thenComposeAsync(fn, executor)
 
-    override def whenComplete(fn: BiConsumer[_ >: T, _ >: Throwable]): CompletableFuture[T] = whenCompleteAsync(fn)
+    override def whenComplete(fn: BiConsumer[_ >: T, _ >: Throwable]): CompletableFuture[T] = whenCompleteAsync(fn, executor)
 
-    override def handle[U](fn: BiFunction[_ >: T, Throwable, _ <: U]): CompletableFuture[U] = handleAsync(fn)
+    override def handle[U](fn: BiFunction[_ >: T, Throwable, _ <: U]): CompletableFuture[U] = handleAsync(fn, executor)
 
     override def exceptionally(fn: JF[Throwable, _ <: T]): CompletableFuture[T] = {
       val cf = new CompletableFuture[T]
@@ -71,7 +73,7 @@ object FuturesConvertersImpl {
             if (n ne this) cf.complete(n.asInstanceOf[T])
           }
         }
-      })
+      }, executor)
       cf
     }