Implementation of a Simple State Machine in Scala/Akka

gistfile1.md

Implementing a Simple State Machine with Akka

Problem

Prof. Charles Pinter's A Book of Abstract Algebra presents the following exercise:

Ch 6 (Functions) part H, problem 4
Given a stream of 1's and 0's, draw the state machine diagram if the stream ends with 111

I came up with the following diagram:

     |  input
     | 0  | 1
state  --------
 s0  | s0 | s1 
 s1  | s0 | s2
 s2  | s0 | s3
 s3  | s0 | s3

The state machine begins in state s0. If it receives a 0, then it stays in that state. Otherwise, it moves to the s1 state. Similarly, s1 and s2 proceed to the next state if the machine receives a 1. Otherwise, it returns to the state0 state.

Consider each state:

• s0 - initial
• s1 - received a single 1
• s2 - received two consecutive 1's
• s3 - received at least three consecutive 1's

Implementing with Akka

I implemented the above state machine using the Akka ActorContext#become method.

We set the Actor's receive method to an initial state. Then, we change its behavior, i.e. how it responds to messages, via the ActorContext#become method. This actor represents the state machine per the above problem.

Let's consider the first state, i.e. s0's implementation in Scala:

class CounterActor extends Actor with ActorLogging {

   import CounterActor._     // imported for `makeErrorMsg`, written to re-use error message code
   import Common._           // object that contains algebraic data types 
                             // Please see this StackOverflow [question](http://stackoverflow.com/questions/26689951/writing-algebraic-data-type-in-scala) for details on ADT's

   override def receive =
     s0                     // Define initial state of Actor

   def s0: PartialFunction[Any, Unit] = {
   	case ONE             => context.become(s1)  // move to `s1` state upon receiving a One
   	case ZERO            => ()                  // do nothing upon receipt of a Zero
   	case TerminateStream => sender ! State0     // reply to whomever sent this messag, i.e. TerminateStream,
   	                                            //  that the current state is 'State0'
   	case badInput        => log.error(makeErrorMsg(badInput)) 
   }

Consider the other classes under the net.fsm package:

• Common.scala - message types for the CounterActor#receive method, as well as types representing the machine's states
• CounterActor.scala - defines the state machine's behavior for each state
• Main.scala - wrapper object that sends messages to an ActorRef or terminates the Stream to (1) learn of its state and (2) reset it

Test in REPL

Let's run a sample in the REPL.

>cd binary_state_machine # name of project
>sbt
>console
 
# import the `CounterActor`
scala> import net.fsm._
import net.fsm._
 
# get the necessary objects for making an ActorSystem
# note - the `Props` import might be unnecessary 
scala> import akka.actor.{Props, ActorSystem}
import akka.actor.{Props, ActorSystem}
 
# Create a new ActorSystem 
scala> val system = ActorSystem("BinaryStream")
system: akka.actor.ActorSystem = akka://BinaryStream
 
# Add an Actor to the ActorSystem 
scala> val actor = system.actorOf(Props[CounterActor], "CounterActor")
actor: akka.actor.ActorRef = Actor[akka://BinaryStream/user/CounterActor#-1803471522]

# Convenience `val`'s to represent a binary 1 and zero 
scala> val one = Common.ONE
one: net.fsm.Common.ONE.type = ONE
 
scala> val zero = Common.ZERO
zero: net.fsm.Common.ZERO.type = ZERO
 
# Send a "1" to the state machine actor 
scala> Main.addToStream(one, actor)
 
# Then, terminate the stream, and get back the machine's current state
# State1 makes sense since, by sending a 1, we moved from s0 -> s1
scala> Main.terminateStream(actor)
res2: Either[net.fsm.Main.FailedStreamTermination,net.fsm.Common.State] = Right(State1)
 
# Terminate the Stream. Observe that the machine's currently in state0.
# This makes sense since, by calling `Main.terminateStream` in the last REPL command,
# the state of the CounterActor was reset to state0.
scala> Main.terminateStream(actor)
res4: Either[net.fsm.Main.FailedStreamTermination,net.fsm.Common.State] = Right(State0)
 
# Now add two 1's consecutively 
scala> Main.addToStream(one, actor)
 
scala> Main.addToStream(one, actor)

# Terminate the stream, observing that the machine is in state2. This makes sense,
# since sending two 1's moves the machine's state from s0 -> s1 -> s2
scala> Main.terminateStream(actor)
res7: Either[net.fsm.Main.FailedStreamTermination,net.fsm.Common.State] = Right(State2)
 
# Next, let's add two 0's to the stream, consecutively. 
scala> Main.addToStream(zero, actor)
 
scala> Main.addToStream(zero, actor)
 
# Let's terminate the stream. It's expected to be in State0 since
# no 1's were added to the stream.
scala> Main.terminateStream(actor)
res10: Either[net.fsm.Main.FailedStreamTermination,net.fsm.Common.State] = Right(State0)
 
# Finally, let's add three 1's consecutively.
scala> Main.addToStream(one, actor)
 
scala> Main.addToStream(one, actor)
 
scala> Main.addToStream(one, actor)
 
# The machine gets terminated, which results in state3 to be printed out.
scala> Main.terminateStream(actor)
res14: Either[net.fsm.Main.FailedStreamTermination,net.fsm.Common.State] = Right(State3)

# stop the actor system, per [docs](http://doc.akka.io/api/akka/2.0/akka/actor/ActorSystem.html)
scala> system.shutdown

Conclusion

This example demonstrated how to use ActorContext#become to implement a state machine.

As an aside, I recommend Dr. Pinter's book for folks interested in Abstract Algebra.

Full code - https://github.com/kevinmeredith/binary_state_machine