SI-7861 Don't execute internal callbacks on the user Executor

Callbacks internal to the implementation of Futures should be
executed with the `InternalCallbackExecutor`, rather than the
user supplied `Executor`.

In a refactoring da54f34a6, `recoverWith` and `flatMap` no longer
played by these rules. This was noticed by a persnickety test in
Play.

Before this patch, the enclosed test outputs:

    % scala-hash v2.10.3-RC2 test/files/run/future-flatmap-exec-count.scala
    mapping
    execute()
    flatmapping
    execute()
    execute()
    recovering
    execute()
    execute()

2 files changed, 67 insertions, 0 deletions

diff --git a/test/files/run/future-flatmap-exec-count.check b/test/files/run/future-flatmap-exec-count.check
new file mode 100644
index 0000000000..dd9dce64ed
--- /dev/null
+++ b/test/files/run/future-flatmap-exec-count.check
@@ -0,0 +1,6 @@
+mapping
+execute()
+flatmapping
+execute()
+recovering
+execute()
diff --git a/test/files/run/future-flatmap-exec-count.scala b/test/files/run/future-flatmap-exec-count.scala
new file mode 100644
index 0000000000..86c37be938
--- /dev/null
+++ b/test/files/run/future-flatmap-exec-count.scala
@@ -0,0 +1,61 @@
+import scala.concurrent._
+import java.util.concurrent.atomic.AtomicInteger
+
+object Test {
+  def main(args: Array[String]) {
+    test()
+  }
+
+  def test() = {
+    def await(f: Future[Any]) =
+      Await.result(f, duration.Duration.Inf)
+
+    val ec = new TestExecutionContext(ExecutionContext.Implicits.global)
+
+    {
+      val p = Promise[Int]()
+      val fp = p.future
+      println("mapping")
+      val mapped = fp.map(x => x)(ec)
+      p.success(0)
+      await(mapped)
+    }
+
+    {
+      println("flatmapping")
+      val p = Promise[Int]()
+      val fp = p.future
+      val flatMapped = fp.flatMap({ (x: Int) =>
+        Future.successful(2 * x)
+      })(ec)
+      p.success(0)
+      await(flatMapped)
+    }
+
+    {
+      println("recovering")
+      val recovered = Future.failed(new Throwable()).recoverWith {
+        case _ => Future.successful(2)
+      }(ec)
+      await(recovered)
+    }
+  }
+
+  class TestExecutionContext(delegate: ExecutionContext) extends ExecutionContext {
+    def execute(runnable: Runnable): Unit = ???
+
+    def reportFailure(t: Throwable): Unit = ???
+
+    override def prepare(): ExecutionContext = {
+      val preparedDelegate = delegate.prepare()
+      return new ExecutionContext {
+        def execute(runnable: Runnable): Unit = {
+          println("execute()")
+          preparedDelegate.execute(runnable)
+        }
+
+        def reportFailure(t: Throwable): Unit = ???
+      }
+    }
+  }
+}