More efficient round trip conversions for Futures

Makes `toJava.toScala` and `toScala.toJava` a wrap-and-unwrap
operation, rather than a wrap-and-rewrap.

Once 2.12.0-M3 is released, we can reduce the overhead by extending
`DefaultPromise` rather than `Promise`.

See:
  4faeac5
  scala/scala#4690

Author: retronym

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

@@ -3,6 +3,7 @@
  */
 package scala.compat.java8
 
+import scala.concurrent.impl.AbstractPromise
 import scala.concurrent.java8.FuturesConvertersImpl._
 import scala.concurrent.{ Future, Promise, ExecutionContext, ExecutionContextExecutorService, ExecutionContextExecutor, impl }
 import java.util.concurrent.{ CompletionStage, Executor, ExecutorService, CompletableFuture }
@@ -54,10 +55,14 @@ object FutureConverters {
    * not support the CompletableFuture interface
    */
   def toJava[T](f: Future[T]): CompletionStage[T] = {
-    val cf = new CF[T]
-    implicit val ec = InternalCallbackExecutor
-    f onComplete cf
-    cf
+    f match {
+      case p: P[T] => p.wrapped
+      case _ =>
+        val cf = new CF[T](f)
+        implicit val ec = InternalCallbackExecutor
+        f onComplete cf
+        cf
+    }
   }
 
   /**
@@ -71,15 +76,13 @@ object FutureConverters {
    * @return a Scala Future that represents the CompletionStage's completion
    */
   def toScala[T](cs: CompletionStage[T]): Future[T] = {
-    val p = Promise[T]()
-    val bc = new BiConsumer[T, Throwable] {
-      override def accept(v: T, e: Throwable): Unit = {
-        if (e == null) p.complete(Success(v))
-        else p.complete(Failure(e))
-      }
+    cs match {
+      case cf: CF[T] => cf.wrapped
+      case _ =>
+        val p = new P[T](cs)
+        cs whenComplete p
+        p.future
     }
-    cs whenComplete bc
-    p.future
   }
 
   /**

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

@@ -5,16 +5,18 @@ package scala.concurrent.java8
 
 // Located in this package to access private[concurrent] members
 
-import scala.concurrent.{ Future, Promise, ExecutionContext, ExecutionContextExecutorService, ExecutionContextExecutor, impl }
+import scala.concurrent._
 import java.util.concurrent.{ CompletionStage, Executor, ExecutorService, CompletableFuture }
+import scala.concurrent.duration.Duration
+import scala.concurrent.impl.Promise.DefaultPromise
 import scala.util.{ Try, Success, Failure }
 import java.util.function.{ BiConsumer, Function ⇒ JF, Consumer, BiFunction }
 
 // TODO: make thie private[scala] when genjavadoc allows for that.
 object FuturesConvertersImpl {
   def InternalCallbackExecutor = Future.InternalCallbackExecutor
 
-  class CF[T] extends CompletableFuture[T] with (Try[T] => Unit) {
+  class CF[T](val wrapped: Future[T]) extends CompletableFuture[T] with (Try[T] => Unit) {
     override def apply(t: Try[T]): Unit = t match {
       case Success(v) ⇒ complete(v)
       case Failure(e) ⇒ completeExceptionally(e)
@@ -78,4 +80,37 @@ object FuturesConvertersImpl {
 
     override def toString: String = super[CompletableFuture].toString
   }
+
+  // TODO Extend DefaultPromise once 2.12.0-M3 is released, and remove `delegate` field and methods that
+  //      refer to it.
+  class P[T](val wrapped: CompletionStage[T]) extends Promise[T] with Future[T] with BiConsumer[T, Throwable] {
+    private val delegate = Promise[T]().asInstanceOf[DefaultPromise[T]]
+
+    override def future: Future[T] = this
+
+    override def tryComplete(result: Try[T]): Boolean = delegate.tryComplete(result)
+
+    override def isCompleted: Boolean = delegate.isCompleted
+
+
+    override def onSuccess[U](pf: PartialFunction[T, U])(implicit executor: ExecutionContext): Unit = super.onSuccess(pf)
+
+
+    override def accept(v: T, e: Throwable): Unit = {
+      if (e == null) complete(Success(v))
+      else complete(Failure(e))
+    }
+
+    override def onComplete[U](f: (Try[T]) => U)(implicit executor: ExecutionContext): Unit = ???
+
+    override def value: Option[Try[T]] = delegate.value
+
+    @throws[Exception](classOf[Exception])
+    override def result(atMost: Duration)(implicit permit: CanAwait): T = delegate.result(atMost)(permit)
+
+    @throws[InterruptedException](classOf[InterruptedException])
+    @throws[TimeoutException](classOf[TimeoutException])
+    override def ready(atMost: Duration)(implicit permit: CanAwait): P.this.type = {delegate.ready(atMost)(permit); this}
+  }
+
 }

src/test/java/scala/compat/java8/FutureConvertersTest.java

@@ -14,6 +14,7 @@
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.Assert.*;
+import static org.junit.Assert.assertSame;
 import static scala.compat.java8.FutureConverters.*;
 
 public class FutureConvertersTest {
@@ -341,4 +342,22 @@ public void testToJavaToCompletableFutureDoesNotMutateUnderlyingPromise() throws
         assertFalse(sf.isCompleted());
         assertFalse(p.isCompleted());
     }
+
+    @Test
+    public void testToJavaAndBackAvoidsWrappers() {
+        final Promise<String> p = promise();
+        final Future<String> sf = p.future();
+        final CompletionStage<String> cs = toJava(sf);
+        Future<String> sf1 = toScala(cs);
+        assertSame(sf, sf1);
+    }
+
+    @Test
+    public void testToScalaAndBackAvoidsWrappers() {
+        final CompletableFuture<String> cf = new CompletableFuture<>();
+        final Future<String> f = toScala(cf);
+        CompletionStage<String> cs1 = toJava(f);
+        assertSame(cf, cs1);
+
+    }
 }