SizeBasedThrottler

sealed trait State
case object Waiting extends State
case object Active extends State

case class Msg(a: Int)
case object Flush

case class StateData(queue: immutable.Queue[Msg])

class SizeBasedThrottler extends FSM[State, StateData] {
startWith(Waiting, StateData(Queue.empty))

onTransition {
case Waiting -> Active =>
nextStateData match {
case StateData(queue) =>
for(x <- queue) yield println(s"$curTime processing ${x.a} ")
Thread.sleep(2000L) // used just to depict as real time problem take time to process request
}
}

when(Active) {
case Event(msg: Msg, _) =>
println(s"$curTime at Active $msg" )
goto(Waiting) using StateData(Queue(msg))//StateData.single(msg)
}

when(Waiting, stateTimeout = 2 seconds){
case Event(msg: Msg, StateData(oldQueue)) =>
val newQueue = oldQueue :+ msg
println(s"$curTime at Idle $newQueue")
stay() using StateData(newQueue)

case Event(Flush, StateData(queue)) => goto(Active) using StateData(queue)

case Event(StateTimeout, StateData(queue)) => goto(Active) using StateData(queue)

}

initialize()
}