Contents

module Workers
  class Worker
    include Workers::Helpers

    def initialize(options = {})
      @logger = Workers::LogProxy.new(options[:logger])
      @input_queue = options[:input_queue] || Queue.new

      @thread = Thread.new { start_event_loop }
    end

    def enqueue(command, data = nil)
      @input_queue.push(Event.new(command, data))

      return nil
    end

    def perform(&block)
      enqueue(:perform, block)

      return nil
    end

    def shutdown(&block)
      enqueue(:shutdown, block)

      return nil
    end

    def join(max_wait = nil)
      raise "Worker can't join itself." if @thread == Thread.current

      return true if !@thread.join(max_wait).nil?

      @thread.kill and return false
    end

    def alive?
      return @thread && @thread.alive?
    end

    private

    def start_event_loop
      while true
        event = @input_queue.pop

        case event.command
        when :shutdown
          shutdown_handler(event)
          return nil
        when :perform
          perform_handler(event)
        else
          process_event(event)
        end
      end
    rescue Exception => e
      exception_handler(e)
    end

    def shutdown_handler(event)
      try_callback(event.data)
    end

    def perform_handler(event)
      try_callback(event.data)
    end

    def exception_handler(e)
      puts concat_e('Worker event loop died.', e)
    end

    def try_callback(callback, &block)
      begin
        callback.call
      rescue Exception => e
        block.call(e) if block
      end
    end

    def process_event(event)
      raise "Unhandled event (#{event.inspect}). Subclass and override if you need custom events."
    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
workers-0.0.7	lib/workers/worker.rb
workers-0.0.6	lib/workers/worker.rb
workers-0.0.5	lib/workers/worker.rb