Contents

module Mutant
  module Parallel
    # Parallel execution worker
    class Worker
      include Adamantium::Flat, Anima.new(
        :mailbox,
        :parent,
        :processor
      )

      # Run worker
      #
      # @param [Hash<Symbol, Object] attributes
      #
      # @return [self]
      def self.run(attributes)
        new(attributes).run
        self
      end

      private_class_method :new

      # Worker loop
      #
      # @return [self]
      #
      # rubocop:disable Lint/Loop
      def run
        begin
          parent.call(Actor::Message.new(:ready, mailbox.sender))
        end until handle(mailbox.receiver.call)
      end

    private

      # Handle job
      #
      # @param [Message] message
      #
      # @return [Boolean]
      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]
      def handle_job(job)
        result = processor.call(job.payload)

        parent.call(
          Actor::Message.new(
            :result,
            JobResult.new(
              job:     job,
              payload: result
            )
          )
        )
      end

    end # Worker
  end # Parallel
end # Mutant

Version data entries

6 entries across 6 versions & 1 rubygems

Version	Path
mutant-0.8.16	lib/mutant/parallel/worker.rb
mutant-0.8.15	lib/mutant/parallel/worker.rb
mutant-0.8.14	lib/mutant/parallel/worker.rb
mutant-0.8.13	lib/mutant/parallel/worker.rb
mutant-0.8.12	lib/mutant/parallel/worker.rb
mutant-0.8.11	lib/mutant/parallel/worker.rb