Contents

require_relative 'batch_execution'

module Upperkut
  class Processor
    def initialize(manager)
      @manager  = manager
      @worker   = @manager.worker
      @logger   = @manager.logger
      @strategy = @worker.strategy

      @sleeping_time = 0
    end

    def run
      @thread ||= Thread.new do
        begin
          process
        rescue Exception => e
          @logger.debug(
            action: :processor_killed,
            reason: e
          )

          @manager.notify_killed_processor(self)
        end
      end
    end

    def kill
      return unless @thread

      @thread.raise Upperkut::Shutdown
      @thread.value # wait
    end

    private

    def process
      loop do
        next if @manager.stopped

        if @strategy.process?
          @sleeping_time = 0
          process_batch
          next
        end

        @sleeping_time += sleep(@worker.setup.polling_interval)
        @logger.debug(sleeping_time: @sleeping_time)
      end
    end

    def process_batch
      BatchExecution.new(@worker, @logger).execute
    end
  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
upperkut-0.7.1	lib/upperkut/processor.rb