class Async::Reactor
def run(*args, &block)
def run(*args, &block) @stopped = false # Allow the user to kick of the initial async tasks. async(*args, &block) if block_given? @timers.wait do |interval| # - nil: no timers # - -ve: timers expired already # - 0: timers ready to fire # - +ve: timers waiting to fire interval = 0 if interval && interval < 0 # Async.logger.debug "[#{self} Pre] Updating #{@fibers.count} fibers..." # Async.logger.debug @fibers.collect{|fiber| [fiber, fiber.alive?]}.inspect # As timeouts may have been updated, and caused fibers to complete, we should check this. @fibers.delete_if{|fiber| !fiber.alive?} # If there is nothing to do, then finish: # Async.logger.debug "[#{self}] @fibers.empty? = #{@fibers.empty?} && interval #{interval.inspect}" return if @fibers.empty? && interval.nil? # Async.logger.debug "Selecting with #{@fibers.count} fibers interval = #{interval}..." if monitors = @selector.select(interval) monitors.each do |monitor| if task = monitor.value # Async.logger.debug "Resuming task #{task} due to IO..." task.resume end end end end until @stopped ensure # Async.logger.debug "[#{self} Ensure] Exiting run-loop (stopped: #{@stopped} exception: #{$!})..." # Async.logger.debug @fibers.collect{|fiber| [fiber, fiber.alive?]}.inspect @stopped = true end