class EventMachine::Reactor

@private

Constants

HeartbeatInterval

Attributes

current_loop_time[R]

Public Class Methods

new() click to toggle source
# File lib/em/pure_ruby.rb, line 278
def initialize
  initialize_for_run
end

Public Instance Methods

add_selectable(io) click to toggle source
# File lib/em/pure_ruby.rb, line 303
def add_selectable io
  @selectables[io.uuid] = io
end
close_loopbreaker() click to toggle source
# File lib/em/pure_ruby.rb, line 406
def close_loopbreaker
  @loopbreak_writer.close
  @loopbreak_writer = nil
end
crank_selectables() click to toggle source
# File lib/em/pure_ruby.rb, line 360
def crank_selectables
  #$stderr.write 'R'

  readers = @selectables.values.select {|io| io.select_for_reading?}
  writers = @selectables.values.select {|io| io.select_for_writing?}

  s = select( readers, writers, nil, @timer_quantum)

  s and s[1] and s[1].each {|w| w.eventable_write }
  s and s[0] and s[0].each {|r| r.eventable_read }

  @selectables.delete_if {|k,io|
    if io.close_scheduled?
      io.close
      true
    end
  }
end
get_selectable(uuid) click to toggle source
# File lib/em/pure_ruby.rb, line 307
def get_selectable uuid
  @selectables[uuid]
end
initialize_for_run() click to toggle source

Called before run, this is a good place to clear out arrays with cruft that may be left over from a previous run. @private

# File lib/em/pure_ruby.rb, line 293
def initialize_for_run
  @running = false
  @stop_scheduled = false
  @selectables ||= {}; @selectables.clear
  @timers = SortedSet.new # []
  set_timer_quantum(0.1)
  @current_loop_time = Time.now
  @next_heartbeat = @current_loop_time + HeartbeatInterval
end
install_oneshot_timer(interval) click to toggle source
# File lib/em/pure_ruby.rb, line 282
def install_oneshot_timer interval
  uuid = UuidGenerator::generate
  #@timers << [Time.now + interval, uuid]
  #@timers.sort! {|a,b| a.first <=> b.first}
  @timers.add([Time.now + interval, uuid])
  uuid
end
open_loopbreaker() click to toggle source
# File lib/em/pure_ruby.rb, line 385
def open_loopbreaker
  # Can't use an IO.pipe because they can't be set nonselectable in Windows.
  # Pick a random localhost UDP port.
  #@loopbreak_writer.close if @loopbreak_writer
  #rd,@loopbreak_writer = IO.pipe
  @loopbreak_reader = UDPSocket.new
  @loopbreak_writer = UDPSocket.new
  bound = false
  100.times {
    @loopbreak_port = rand(10000) + 40000
    begin
      @loopbreak_reader.bind "localhost", @loopbreak_port
      bound = true
      break
    rescue
    end
  }
  raise "Unable to bind Loopbreaker" unless bound
  LoopbreakReader.new(@loopbreak_reader)
end
run() click to toggle source
# File lib/em/pure_ruby.rb, line 311
def run
  raise Error.new( "already running" ) if @running
  @running = true

  begin
    open_loopbreaker

    loop {
      @current_loop_time = Time.now

      break if @stop_scheduled
      run_timers
      break if @stop_scheduled
      crank_selectables
      break if @stop_scheduled
      run_heartbeats
    }
  ensure
    close_loopbreaker
    @selectables.each {|k, io| io.close}
    @selectables.clear

    @running = false
  end

end
run_heartbeats() click to toggle source
# File lib/em/pure_ruby.rb, line 353
def run_heartbeats
  if @next_heartbeat <= @current_loop_time
    @next_heartbeat = @current_loop_time + HeartbeatInterval
    @selectables.each {|k,io| io.heartbeat}
  end
end
run_timers() click to toggle source
# File lib/em/pure_ruby.rb, line 338
def run_timers
  @timers.each {|t|
    if t.first <= @current_loop_time
      @timers.delete t
      EventMachine::event_callback "", TimerFired, t.last
    else
      break
    end
  }
  #while @timers.length > 0 and @timers.first.first <= now
  #  t = @timers.shift
  #  EventMachine::event_callback "", TimerFired, t.last
  #end
end
set_timer_quantum(interval_in_seconds) click to toggle source
# File lib/em/pure_ruby.rb, line 416
def set_timer_quantum interval_in_seconds
  @timer_quantum = interval_in_seconds
end
signal_loopbreak() click to toggle source
# File lib/em/pure_ruby.rb, line 411
def signal_loopbreak
  #@loopbreak_writer.write '+' if @loopbreak_writer
  @loopbreak_writer.send('+',0,"localhost",@loopbreak_port) if @loopbreak_writer
end
stop() click to toggle source

stop

# File lib/em/pure_ruby.rb, line 380
def stop
  raise Error.new( "not running") unless @running
  @stop_scheduled = true
end