Contents

module Larva
  class WorkerPool
    def self.start(processors)
      new(processors).start
    end

    attr_reader :processors, :workers
    def initialize(processors)
      @processors = processors
    end

    def start
      start_workers
      keep_workers_alive if workers.count > 0
    end

    private
    def start_workers
      logger.info "Starting #{processors.count} threads."
      @workers = processors.map do |topic, processor|
        worker = Thread.new { start_worker(topic, processor) }
        worker[:name] = "Listener for #{topic}"
        worker
      end
      logger.info "#{workers.count} threads started."
    end

    def start_worker(topic, processor)
      Larva::Listener.listen(topic, processor)
    rescue => e
      logger.error "Unexpected listener termination: #{e} #{e.backtrace}"
    end

    def keep_workers_alive
      sleep(5) while workers.all? { |t| t.alive?  }
      logger.error 'Some threads have died:'
      workers.each do |worker|
        logger.error "#{worker[:name]} was #{worker.alive? ? 'alive' : 'dead'}"
      end
    end

    def logger
      Propono.config.logger
    end
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
larva-0.7.1	lib/larva/worker_pool.rb
larva-0.7.0	lib/larva/worker_pool.rb