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
|
module Celluloid
class Mailbox
# An alternative implementation of Celluloid::Mailbox using Reactor
class Evented < Celluloid::Mailbox
attr_reader :reactor
def initialize(reactor_class)
super()
# @condition won't be used in the class.
@reactor = reactor_class.new
end
# Add a message to the Mailbox
def <<(message)
@mutex.lock
begin
if mailbox_full || @dead
dead_letter(message)
return
end
if message.is_a?(SystemEvent)
# SystemEvents are high priority messages so they get added to the
# head of our message queue instead of the end
@messages.unshift message
else
@messages << message
end
ensure
begin
@mutex.unlock
rescue
nil
end
end
begin
current_actor = Thread.current[:celluloid_actor]
@reactor.wakeup unless current_actor && current_actor.mailbox == self
rescue => ex
Internals::Logger.crash "reactor crashed", ex
dead_letter(message)
end
nil
end
# Receive a message from the Mailbox
def check(timeout = nil, &block)
# Get a message if it is available and process it immediately if possible:
if message = next_message(block)
return message
end
# ... otherwise, run the reactor once, either blocking or will return
# after the given timeout:
@reactor.run_once(timeout)
# No message was received:
nil
end
# Obtain the next message from the mailbox that matches the given block
def next_message(block)
@mutex.lock
begin
super(&block)
ensure
begin
@mutex.unlock
rescue
nil
end
end
end
# Cleanup any IO objects this Mailbox may be using
def shutdown
super do
@reactor.shutdown
end
end
end
end
end
|
