Class: Celluloid::Mailbox::Evented

Inherits:
Celluloid::Mailbox show all
Defined in:
lib/celluloid/mailbox/evented.rb

Overview

An alternative implementation of Celluloid::Mailbox using Reactor

Instance Attribute Summary collapse

Attributes inherited from Celluloid::Mailbox

#address, #max_size

Instance Method Summary collapse

Methods inherited from Celluloid::Mailbox

#alive?, #each, #inspect, #receive, #size, #to_a

Constructor Details

#initialize(reactor_class) ⇒ Evented

Returns a new instance of Evented.



7
8
9
10
11
 
# File 'lib/celluloid/mailbox/evented.rb', line 7

def initialize(reactor_class)
  super()
  # @condition won't be used in the class.
  @reactor = reactor_class.new
end

Instance Attribute Details

#reactor ⇒ Object (readonly)

Returns the value of attribute reactor.



5
6
7
 
# File 'lib/celluloid/mailbox/evented.rb', line 5

def reactor
  @reactor
end

Instance Method Details

#<<(message) ⇒ Object

Add a message to the Mailbox



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
 
# File 'lib/celluloid/mailbox/evented.rb', line 14

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
    @mutex.unlock rescue nil
  end
  begin
    current_actor = Thread.current[:celluloid_actor]
    @reactor.wakeup unless current_actor && current_actor.mailbox == self
  rescue
    Internals::Logger.crash "reactor crashed", $ERROR_INFO
    dead_letter(message)
  end
  nil
end

#check(timeout = nil, &block) ⇒ Object

Receive a message from the Mailbox



42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/celluloid/mailbox/evented.rb', line 42

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:
  return nil
end

#next_message(block) ⇒ Object

Obtain the next message from the mailbox that matches the given block



57
58
59
60
61
62
63
64
 
# File 'lib/celluloid/mailbox/evented.rb', line 57

def next_message(block)
  @mutex.lock
  begin
    super(&block)
  ensure
    @mutex.unlock rescue nil
  end
end

#shutdown ⇒ Object

Cleanup any IO objects this Mailbox may be using



67
68
69
70
71
 
# File 'lib/celluloid/mailbox/evented.rb', line 67

def shutdown
  super do
    @reactor.shutdown
  end
end