More efficient round trip conversions for Futures

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

This won't be possible if we merge scala#46, which speaks in favour
of reverting the change that finalized `DefaultPromise` in Scala
2.12.x.

This patch does not attempt to "wrap the Future with an implementation
of CompletionStage that delegates to Future", as Jame's proposed in
issue scala#44, which seems tricky or at least tedious to implement.

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,9 +76,13 @@ object FutureConverters {
    * @return a Scala Future that represents the CompletionStage's completion
    */
   def toScala[T](cs: CompletionStage[T]): Future[T] = {
-    val p = new P[T]
-    cs whenComplete p
-    p.future
+    cs match {
+      case cf: CF[T] => cf.wrapped
+      case _ =>
+        val p = new P[T](cs)
+        cs whenComplete p
+        p.future
+    }
   }
 
   /**

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

@@ -6,15 +6,15 @@ package scala.concurrent.java8
 // Located in this package to access private[concurrent] members
 
 import scala.concurrent.{ Future, Promise, ExecutionContext, ExecutionContextExecutorService, ExecutionContextExecutor, impl }
-import java.util.concurrent.{ CompletionStage, Executor, ExecutorService, CompletableFuture }
+import java.util.concurrent._
 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)
@@ -79,7 +79,7 @@ object FuturesConvertersImpl {
     override def toString: String = super[CompletableFuture].toString
   }
 
-  class P[T] extends impl.Promise.DefaultPromise[T] with BiConsumer[T, Throwable] {
+  class P[T](val wrapped: CompletionStage[T]) extends impl.Promise.DefaultPromise[T] with BiConsumer[T, Throwable] {
     override def accept(v: T, e: Throwable): Unit = {
       if (e == null) complete(Success(v))
       else complete(Failure(e))

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);
+
+    }
 }