Akka.io - Pi calculating actors example ported to JRuby

pi.rb

require 'java'
require 'mvn:org.scala-lang:scala-library'
require 'case'

module Akka
  include_package 'akka.actor'
  include_package 'akka.util.duration'
  java_import 'akka.routing.RoundRobinRouter'
  java_import 'akka.util.Duration'
end

module Pi
  class PiMessage
  end

  class Calculate < PiMessage
  end

  class Work < PiMessage
    attr_reader :start, :nrOfElements
    def initialize(start, nrOfElements)
      @start = start
      @nrOfElements = nrOfElements
    end
  end

  class Result < PiMessage
    attr_reader :value
    def initialize(value)
      @value = value
    end
  end

  class PiApproximation
    attr_reader :pi, :duration
    def initialize(pi, duration)
      @pi = pi
      @duration = duration
    end
  end
end

class Worker < Akka::UntypedActor
  def self.create(*args)
    self.new(*args)
  end

  def calculatePiFor(start, nrOfElements)
    acc = 0.0
    (start..(start + nrOfElements)).each do |i|
      acc = acc + 4.0 * (1 - (i % 2) * 2) / (2 * i + 1)
    end
    acc
  end

  def onReceive(message)
    case message
    when Case::All[Pi::Work]
      sender.tell(Pi::Result.new(calculatePiFor(message.start, message.nrOfElements)), getSelf)
    when Case::All[Object]
      puts "Unknown message!"
    end
  end
end

class Master < Akka::UntypedActor
  def self.create(*args)
    self.new(*args)
  end

  def initialize(nrOfWorkers, nrOfMessages, nrOfElements, listener)
    super()
    @nrOfMessages = nrOfMessages
    @nrOfElements = nrOfElements
    @listener = listener
    @start = Java::JavaLang::System.currentTimeMillis
    @workerRouter = context.actorOf(Akka::Props.new(Worker).withRouter(Akka::RoundRobinRouter.new(nrOfWorkers)),
                                    name = "workerRouter")
    @pi = 0
    @nrOfResults = 0
  end

  def onReceive(message)
    case message
    when Case::All[Pi::Calculate]
      (0...@nrOfMessages).each do |i|
        @workerRouter.send(:"$bang", Pi::Work.new(i * @nrOfElements, @nrOfElements), getSelf)
      end
    when Case::All[Pi::Result]
      @pi += message.value
      @nrOfResults += 1
      if (@nrOfResults == @nrOfMessages)
        @listener.tell(Pi::PiApproximation.new(@pi, Java::JavaLang::System.currentTimeMillis - @start), getSelf)
        context.stop(getSelf)
      end
    when Case::All[Object]
      puts "Unknown message received!!!"
    end
  end
end

class Listener < Akka::UntypedActor
  def self.create(*args)
    self.new(*args)
  end


  def onReceive(message)
    case message
    when Case::All[Pi::PiApproximation]
      puts "**** Pi approximation: #{message.pi}. Time taken: #{message.duration} ***"
      context.system.shutdown
    when Case::All[Object]
      puts "Unknown message received in Listener!!!"
    end
  end
end

class MasterFactory
  include Akka::UntypedActorFactory

  def initialize(nrOfWorkers, nrOfMessages, nrOfElements, listener)
    @nrOfWorkers = nrOfWorkers
    @nrOfMessages = nrOfMessages
    @nrOfElements = nrOfElements
    @listener = listener
  end
  def create
    Master.new(@nrOfWorkers, @nrOfMessages, @nrOfElements, @listener)
  end
end

def calculate(nrOfWorkers, nrOfElements, nrOfMessages)
  system = Akka::ActorSystem.apply("PiSystem")
  listener = system.actorOf(Akka::Props.new(Listener), name = "listener")
  master = system.actorOf(Akka::Props.new(MasterFactory.new(nrOfWorkers, nrOfMessages, nrOfElements, listener)), name = "master")
  master.send(:"$bang", Pi::Calculate.new, nil)
end