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
116
117
118
|
/* __ *\
** ________ ___ / / ___ Scala API **
** / __/ __// _ | / / / _ | (c) 2003-2011, LAMP/EPFL **
** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
** /____/\___/_/ |_/____/_/ | | **
** |/ **
\* */
package scala.concurrent
import java.util.concurrent.atomic.{ AtomicInteger }
import java.util.concurrent.{ Executors, Future => JFuture, Callable }
import scala.util.Duration
import scala.concurrent.forkjoin.{ ForkJoinPool, RecursiveTask => FJTask, RecursiveAction, ForkJoinWorkerThread }
import scala.collection.generic.CanBuildFrom
import collection._
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
/* implementations follow */
private implicit val executionContext = this
/** 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 buffer = new mutable.ArrayBuffer[T]
val counter = new AtomicInteger(1) // how else could we do this?
val p: Promise[Coll[T]] = promise[Coll[T]] // we need an implicit execctx in the signature
var idx = 0
def tryFinish() = if (counter.decrementAndGet() == 0) {
val builder = cbf(futures)
builder ++= buffer
p success builder.result
}
for (f <- futures) {
val currentIndex = idx
buffer += null.asInstanceOf[T]
counter.incrementAndGet()
f onComplete {
case Left(t) =>
p tryFailure t
case Right(v) =>
buffer(currentIndex) = v
tryFinish()
}
idx += 1
}
tryFinish()
p.future
}
/** TODO some docs
*
*/
def any[T](futures: Traversable[Future[T]]): Future[T] = {
val p = promise[T]
val completeFirst: Either[Throwable, T] => Unit = elem => p tryComplete elem
futures foreach (_ onComplete completeFirst)
p.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 count = 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 (count.decrementAndGet() == 0) result trySuccess None
}
futures.foreach(_ onComplete search)
result.future
}
}
}
sealed trait CanAwait
|
