free_mutant/lib/mutant/parallel/worker.rb

75 lines
1.5 KiB
Ruby
Raw Normal View History

2014-10-23 07:37:53 -04:00
module Mutant
2014-12-08 19:10:31 -05:00
module Parallel
# Parallel execution worker
2014-10-23 07:37:53 -04:00
class Worker
include Adamantium::Flat, Anima.new(:mailbox, :processor, :parent)
2014-10-23 07:37:53 -04:00
# Run worker
#
# @param [Hash<Symbol, Object] attributes
#
2014-12-08 19:10:31 -05:00
# @return [self]
2014-10-23 07:37:53 -04:00
#
# @api private
#
def self.run(attributes)
2014-12-08 19:10:31 -05:00
new(attributes).run
self
2014-10-23 07:37:53 -04:00
end
2014-12-08 19:10:31 -05:00
private_class_method :new
2014-10-23 07:37:53 -04:00
# Worker loop
#
# @return [self]
#
# @api private
#
# rubocop:disable Lint/Loop
#
2014-12-08 19:10:31 -05:00
def run
2014-10-23 07:37:53 -04:00
begin
parent.call(Actor::Message.new(:ready, mailbox.sender))
end until handle(mailbox.receiver.call)
2014-10-23 07:37:53 -04:00
end
2014-12-08 19:10:31 -05:00
private
2014-10-23 07:37:53 -04:00
# Handle job
#
# @param [Message] message
#
# @return [Boolean]
#
2014-12-01 13:33:47 -05:00
# @api private
#
2014-10-23 07:37:53 -04:00
def handle(message)
type, payload = message.type, message.payload
case message.type
when :job
handle_job(payload)
nil
when :stop
true
else
fail Actor::ProtocolError, "Unknown command: #{type.inspect}"
end
end
# Handle mutation
#
# @param [Job] job
#
# @return [undefined]
#
# @api private
#
def handle_job(job)
2014-12-08 19:10:31 -05:00
result = processor.call(job.payload)
parent.call(Actor::Message.new(:result, JobResult.new(job: job, payload: result)))
2014-10-23 07:37:53 -04:00
end
end # Worker
2014-12-08 19:10:31 -05:00
end # Parallel
2014-10-23 07:37:53 -04:00
end # Mutant