blob: 576d203d8905cc168012ebb623624ab46f990097 (
plain) (
blame)
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
|
package scala.concurrent;
class MailBox extends Monitor {
type Message = AnyRef;
private abstract class Receiver extends Monitor {
type t;
val receiver: PartialFunction[Message, t];
var msg: Message = _;
def receive(): t = synchronized {
if (msg != null) wait();
receiver(msg)
}
def receiveWithin(msec: Long): t = synchronized {
if (msg != null) wait(msec);
receiver(if (msg != null) msg else TIMEOUT)
}
}
private val sent = new LinkedList[Message];
private var lastSent = sent;
private var receivers = new LinkedList[Receiver];
private var lastReceiver = receivers;
def send(msg: Message): Unit = synchronized {
var rs = receivers, rs1 = rs.next;
while (rs1 != null && !rs1.elem.receiver.isDefinedAt(msg)) {
rs = rs1; rs1 = rs1.next;
}
if (rs1 != null) {
rs.next = rs1.next; rs1.elem.msg = msg; rs1.elem.notify();
} else {
lastSent = lastSent.append(msg)
}
}
def scanSentMsgs[a](r: Receiver { type t = a }): Unit = synchronized {
var ss = sent, ss1 = ss.next;
while (ss1 != null && !r.receiver.isDefinedAt(ss1.elem)) {
ss = ss1; ss1 = ss1.next
}
if (ss1 != null) {
ss.next = ss1.next; r.msg = ss1.elem;
} else {
lastReceiver = lastReceiver append r;
}
}
def receive[a](f: PartialFunction[Message, a]): a = {
val r = new Receiver { type t = a; val receiver = f }
scanSentMsgs(r);
r.receive()
}
def receiveWithin[a](msec: Long)(f: PartialFunction[Message, a]): a = {
val r = new Receiver { type t = a; val receiver = f }
scanSentMsgs(r);
r.receiveWithin(msec)
}
}
|
