Contents

module Hawkei
  module Processor
    class Async

      def initialize
        @queue = Queue.new
        @state_worker = Concurrent::AtomicBoolean.new(true)
        @worker = Worker.new(@queue, @state_worker)

        at_exit do
          shutdown_worker if worker_running?
        end
      end

      def enqueue(attributes)
        ensure_worker_running

        @queue << attributes

        true
      end

      private

      def shutdown_worker
        @state_worker.make_false
        @queue << Hawkei::Processor::Worker::SHUTDOWN_MESSAGE
        @worker_thread.wait

        executor.shutdown
        executor.wait_for_termination
      end

      def executor
        @executor ||= Concurrent.global_io_executor
      end

      def ensure_worker_running
        return if worker_running?

        @worker_thread = Concurrent::Future.execute { @worker.run }
      end

      def worker_running?
        @worker_thread && @worker_thread.incomplete?
      end

    end
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
hawkei-1.1.0	lib/hawkei/processor/async.rb
hawkei-1.0.0	lib/hawkei/processor/async.rb