1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
/* __ *\
** ________ ___ / / ___ Scala API **
** / __/ __// _ | / / / _ | (c) 2003-2011, LAMP/EPFL **
** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
** /____/\___/_/ |_/____/_/ | | **
** |/ **
\* */
package scala.concurrent
import java.util.concurrent.{ Executors, Future => JFuture, Callable }
import java.util.concurrent.atomic.{ AtomicInteger }
import scala.util.{ Duration, Timeout }
import scala.concurrent.forkjoin.{ ForkJoinPool, RecursiveTask => FJTask, RecursiveAction, ForkJoinWorkerThread }
import scala.collection.generic.CanBuildFrom
trait ExecutionContext {
protected implicit object CanAwaitEvidence extends CanAwait
def execute(runnable: Runnable): Unit
def execute[U](body: () => U): Unit
def promise[T]: Promise[T]
def future[T](body: Callable[T]): Future[T] = future(body.call())
def future[T](body: => T): Future[T]
def blocking[T](atMost: Duration)(body: =>T): T
def blocking[T](awaitable: Awaitable[T], atMost: Duration): T
def futureUtilities: FutureUtilities = FutureUtilitiesImpl
}
sealed trait CanAwait
trait FutureUtilities {
/** TODO some docs
*
*/
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
}
/** TODO some docs
*
*/
def any[T](futures: Traversable[Future[T]]): Future[T] = {
val futureResult = promise[T]()
val completeFirst: Either[Throwable, T] => Unit = futureElem => futureResult tryComplete futureElem
futures.foreach(_ onComplete completeFirst)
futureResult.future
}
/** TODO some docs
*
*/
def find[T](futures: Traversable[Future[T]])(predicate: T => Boolean): Future[Option[T]] = {
if (futures.isEmpty) Promise.successful[Option[T]](None).future
else {
val result = promise[Option[T]]()
val ref = new AtomicInteger(futures.size)
val search: Either[Throwable, T] ⇒ Unit = {
v ⇒ v match {
case Right(r) ⇒ if (predicate(r)) result trySuccess Some(r)
case _ ⇒
}
if (ref.decrementAndGet == 0) result trySuccess None
}
futures.foreach(_ onComplete search)
result.future
}
}
}
object FutureUtilitiesImpl extends FutureUtilities {
}
|
