Best way to integrate autonomous and reactive behavior in a Scala actor?

Question

I am an experienced Java programmer and I am starting to develop actor-based Scala applications. In the application that I am currently developing, I have to deal with the implementation of the role of the Sender, exhibiting both autonomous and reactive behavior. The scenario is as follows (pseudo-code):

Actor Sender{ Active behavior (must be active once the actor boots): do-in-sequence{ send to Stdout A send to Stdout B send to Stdout C send stop to Stdout and then exit } Reactive behavior (must be active once the actor boots): as soon as received stop from StopNotifier -> send stop to Stdout and then exit } } Actor Stdout{ Purely reactive behavior (ie wait for msg){ as soon as received A -> print A as soon as received B -> print B as soon as received C -> print C as soon as received stop from Sender -> exit } } Actor StopNotifier Purely active behavior { compute, and when some condition is met -> send stop to Sender } 

My question is: is this the best way to express autonomous behavior for a Scala actor who needs to integrate autonomy and reactivity (as described in this article )?

In other words, what is the best way / style to encode the Sender actor in the above example?

I came up with a solution (see below), but since I am not a Scala guru (for now :) :), I would like to know if it is possible to improve what I implemented in an improved / improved solution.

 case object START case object A case object B case object C case object SENT_A case object SENT_B case object ACK_A case object ACK_B case object ACK_C case object STOP class Sender(stdout: Stdout) extends Actor { def act() { self!START while (true){ receive { case START => stdout!?A self!SENT_A case SENT_A => stdout!?B self!SENT_B case SENT_B => stdout!?C stdout!?STOP exit() case STOP => { Console.println("[Sender:]Received STOP, terminating") stdout!?STOP exit() } } } } } class Stdout() extends Actor { def act() { while (true) { receive{ case A => Console.println("[Stdout:]A") reply(ACK_A) case B => Console.println("[Stdout:]B") reply(ACK_B) case C => Console.println("[Stdout:]C") reply(ACK_C) exit() case STOP => Console.println("[Stdout:]Received STOP, terminating") exit() } } } } class StopNotifier(sender: Sender) extends Actor { def act() { /* * The idea is that the StopNotifier should send a STOP message to the Sender * when a certain condition is met. * The sleep used here is just a semplification, since the detection of such * a condition is not relevant for the example. */ Thread.sleep(200) Console.println("[StopNotifier:]Sending STOP to sender") sender ! STOP exit() } } object app extends Application { val stdout = new Stdout stdout.start val sender = new Sender(stdout) sender.start val stopNotifier = new StopNotifier(sender) stopNotifier.start } 

In particular, what bothers me in my current implementation is that in order to quickly respond to receiving a STOP message from StopNotifier, I had to send messages at each stage of the Sender’s execution (i.e., after sending A, B to the actor Stdout). It seems too complicated to be the right way to do things :).

I also tried to develop another solution using other Scala language conventions (e.g. asynchronous sending, reaction, etc.), but somehow they seemed to me affected by other problems / tricks.

Does anyone have a better solution to combat the integration of autonomy and reactive behavior in Scala actors?

Answer 1

If I understand correctly, you should use Akka actors instead of Akka FSM specifically to simulate the sender as a state machine. Akka actors have a built-in stop mechanism or you can always use your own message, which could be processed from all states using the whenUnhandled handler.

See http://doc.akka.io/docs/akka/snapshot/scala/fsm.html

This is clearly overpriced for this, but I assume that you are trying to do something more complex. You can also have Stdout "watch" Sender so that it ends when Sender completes, and not when it receives a specific message. See Life Cycle Monitoring aka DeathWatch .

 package fsmTest import akka.actor._ import akka.util.duration._ sealed trait Msg case object A extends Msg case object B extends Msg case object C extends Msg sealed trait SenderState case object Started extends SenderState case object SentA extends SenderState case object SentB extends SenderState case class SenderData() class Sender(stdout: ActorRef) extends Actor with FSM[SenderState, SenderData] { case object GoNextState startWith(Started, SenderData()) when(Started) { case Event(GoNextState, data) => { stdout ! A goto(SentA) using data } } when(SentA) { case Event(GoNextState, data) => { stdout ! B goto(SentB) using data } } when(SentB) { case Event(GoNextState, data) => { stdout ! C goto(Started) using data } } // //Handle messages which aren't explicitly handled in state here // whenUnhandled { // case Event(SomeCustomStop, data) => { // stop(FSM.Shutdown) // } // } setTimer("goNextState", GoNextState, 1 second, repeat = true) initialize } class Stdout() extends Actor { def receive = { case msg: Msg => { context.watch(sender) //Not sure if you're gonna want to do this here, but you get the point println(msg) } case Terminated(_) => context.stop(self) } } object FSMTest extends App { implicit val system = ActorSystem("Testing") val stdout = system.actorOf(Props[Stdout], "stdout") val sender = system.actorOf(Props(new Sender(stdout)), "sender") system.scheduler.scheduleOnce(5 seconds) { system.stop(sender) system.shutdown() } system.awaitTermination(10 seconds) } 

Regardless of how you implement the state in the sender, if you want to use Actors to simulate it, I believe that you will need to "send" messages either in the processing of events, or using a timer, as I have above.

Answer 2

In this order, messages will be sent, sent in order from one participant within one reaction or reception unit. (You may have other messages from other participants, but you will not send A then B and get B, then A.)

So you can just

 stdout ! A stdout ! B stdout ! C 

unless you need to do something else.