Class: Celluloid::IO::Reactor

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/celluloid/io/reactor.rb

Overview

React to external I/O events. This is kinda sorta supposed to resemble the Reactor design pattern.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Reactor

Returns a new instance of Reactor.



15
16
17
 
# File 'lib/celluloid/io/reactor.rb', line 15

def initialize
  @selector = NIO::Selector.new
end

Instance Method Details

#run_once(timeout = nil) ⇒ Object

Run the reactor, waiting for events or wakeup signal



48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/celluloid/io/reactor.rb', line 48

def run_once(timeout = nil)
  @selector.select(timeout) do |monitor|
    task = monitor.value
    monitor.close

    if task.running?
      task.resume
    else
      Logger.warn("reactor attempted to resume a dead task")
    end
  end
end

#wait(io, set) ⇒ Object

Wait for the given IO operation to complete



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/celluloid/io/reactor.rb', line 30

def wait(io, set)
  # zomg ugly type conversion :(
  unless io.is_a?(::IO) or io.is_a?(OpenSSL::SSL::SSLSocket)
    if io.respond_to? :to_io
      io = io.to_io
    elsif ::IO.respond_to? :try_convert
      io = ::IO.try_convert(io)
    end

    raise TypeError, "can't convert #{io.class} into IO" unless io.is_a?(::IO)
  end

  monitor = @selector.register(io, set)
  monitor.value = Task.current
  Task.suspend :iowait
end

#wait_readable(io) ⇒ Object

Wait for the given IO object to become readable



20
21
22
 
# File 'lib/celluloid/io/reactor.rb', line 20

def wait_readable(io)
  wait io, :r
end

#wait_writable(io) ⇒ Object

Wait for the given IO object to become writable



25
26
27
 
# File 'lib/celluloid/io/reactor.rb', line 25

def wait_writable(io)
  wait io, :w
end