Merge remote branch 'origin/execution-context' into topic/stash

Author: phaller

src/library/scala/concurrent/Awaitable.scala

@@ -11,15 +11,14 @@ package scala.concurrent
 
 
 import scala.annotation.implicitNotFound
-import scala.util.Timeout
+import scala.util.Duration
 
 
 
 trait Awaitable[+T] {
-  @implicitNotFound(msg = "Waiting must be done by calling `await(timeout) b`, where `b` is the `Awaitable` object.")
-  def await(timeout: Timeout)(implicit canblock: CanBlock): T
+  @implicitNotFound(msg = "Waiting must be done by calling `blocking(timeout) b`, where `b` is the `Awaitable` object or a potentially blocking piece of code.")
+  def await(atMost: Duration)(implicit canawait: CanAwait): T
 }
 
 
 
-

src/library/scala/concurrent/ExecutionContext.scala

@@ -1,16 +1,25 @@
+/*                     __                                               *\
+**     ________ ___   / /  ___     Scala API                            **
+**    / __/ __// _ | / /  / _ |    (c) 2003-2011, LAMP/EPFL             **
+**  __\ \/ /__/ __ |/ /__/ __ |    http://scala-lang.org/               **
+** /____/\___/_/ |_/____/_/ | |                                         **
+**                          |/                                          **
+\*                                                                      */
+
 package scala.concurrent
 
 
 
 import java.util.concurrent.{ Executors, Future => JFuture, Callable }
 import scala.util.{ Duration, Timeout }
 import scala.concurrent.forkjoin.{ ForkJoinPool, RecursiveTask => FJTask, RecursiveAction, ForkJoinWorkerThread }
+import scala.collection.generic.CanBuildFrom
 
 
 
 trait ExecutionContext {
 
-  protected implicit object CanBlockEvidence extends CanBlock
+  protected implicit object CanAwaitEvidence extends CanAwait
 
   def execute(runnable: Runnable): Unit
 
@@ -22,12 +31,47 @@ trait ExecutionContext {
 
   def future[T](body: => T): Future[T]
 
-  /** Only callable from the tasks running on the same execution context. */
-  def blockingCall[T](timeout: Timeout, body: Awaitable[T]): T
+  def blocking[T](atMost: Duration)(body: =>T): T
+  
+  def blocking[T](awaitable: Awaitable[T], atMost: Duration): T
+  
+  def futureUtilities: FutureUtilities = FutureUtilitiesImpl
 
 }
 
 
-sealed trait CanBlock
+sealed trait CanAwait
+
+
+trait FutureUtilities {
+  
+  def all[T, Coll[X] <: Traversable[X]](futures: Coll[Future[T]])(implicit cbf: CanBuildFrom[Coll[_], T, Coll[T]]): Future[Coll[T]] = {
+    val builder = cbf(futures)
+    val p: Promise[Coll[T]] = promise[Coll[T]]
+    
+    if (futures.size == 1) futures.head onComplete {
+      case Left(t) => p failure t
+      case Right(v) => builder += v
+        p success builder.result
+    } else {
+      val restFutures = all(futures.tail)
+      futures.head onComplete {
+        case Left(t) => p failure t
+        case Right(v) => builder += v
+          restFutures onComplete {
+            case Left(t) => p failure t
+            case Right(vs) => for (v <- vs) builder += v
+              p success builder.result
+          }
+      }
+    }
+    
+    p.future
+  }
+  
+}
 
 
+object FutureUtilitiesImpl extends FutureUtilities {
+}
+