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
|
package cask.actor
import collection.mutable
abstract class BaseActor[T]()(implicit ac: Context) extends Actor[T]{
private val queue = new mutable.Queue[(T, Context.Token)]()
private var scheduled = false
def send(t: T)
(implicit fileName: sourcecode.FileName,
line: sourcecode.Line): Unit = synchronized{
val token = ac.reportSchedule(this, t, fileName, line)
queue.enqueue((t, token))
if (!scheduled){
scheduled = true
ac.execute(() => runWithItems())
}
}
def sendAsync(f: scala.concurrent.Future[T])
(implicit fileName: sourcecode.FileName,
line: sourcecode.Line) = {
f.onComplete{
case scala.util.Success(v) => this.send(v)
case scala.util.Failure(e) => ac.reportFailure(e)
}
}
def runBatch0(msgs: Seq[(T, Context.Token)]): Unit
private[this] def runWithItems(): Unit = {
val msgs = synchronized(queue.dequeueAll(_ => true))
runBatch0(msgs)
synchronized{
if (queue.nonEmpty) ac.execute(() => runWithItems())
else{
assert(scheduled)
scheduled = false
}
}
}
}
abstract class BatchActor[T]()(implicit ac: Context) extends BaseActor[T]{
def runBatch(msgs: Seq[T]): Unit
def runBatch0(msgs: Seq[(T, Context.Token)]): Unit = {
try {
msgs.foreach{case (m, token) => ac.reportRun(this, m, token)}
runBatch(msgs.map(_._1))
}
catch{case e: Throwable => ac.reportFailure(e)}
finally msgs.foreach{case (m, token) => ac.reportComplete(token)}
}
}
abstract class SimpleActor[T]()(implicit ac: Context) extends BaseActor[T]{
def run(msg: T): Unit
override def runBatch0(msgs: Seq[(T, Context.Token)]): Unit = {
for((msg, token) <- msgs) {
try {
ac.reportRun(this, msg, token)
run(msg)
}
catch{case e: Throwable => ac.reportFailure(e)}
finally ac.reportComplete(token)
}
}
}
abstract class StateMachineActor[T]()(implicit ac: Context) extends SimpleActor[T]() {
class State(run0: T => State = null){
def run = run0
}
protected[this] def initialState: State
protected[this] var state: State = initialState
def run(msg: T): Unit = {
assert(state != null)
state = state.run(msg)
}
}
class ProxyActor[T, V](f: T => V, downstream: Actor[V])
(implicit ac: Context) extends SimpleActor[T]{
def run(msg: T): Unit = downstream.send(f(msg))
}
|
