akkademo-gatekeeper

import akka.actor._
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

class WaitFor(var messages: List[Any => Boolean], andThen: ActorRef) extends Actor with Stash {

  var received = List[Any]()

  val justForward: Receive = { case msg => andThen forward msg}

  def doneWaiting() {
    andThen ! received
    context.become(justForward)
    unstashAll()
  }

  def receive = { case msg =>
    val (matches, doesnt) = messages.partition(fn => fn(msg))
    if (matches.isEmpty)
      stash()
    else {
      received = msg :: received
      messages = doesnt
      if (messages.isEmpty) {
        doneWaiting()
      }
    }
  }
}

trait PrintAll extends Actor {
  def receive = { case msg => println("printing " + msg) }
}

class Main extends Actor {


  override def preStart() = {
    println("starting")
    val p = context.actorOf(Props(new PrintAll {} ))
    val w = context.actorOf(Props(new WaitFor(List(_ == 1, _ == 2, _ == 3, _ == 5), p)))
    for(i <- (1 until 10))
      context.system.scheduler.scheduleOnce(1.second, w, i)

    context.system.scheduler.scheduleOnce(2.second, self, PoisonPill)
    println("started")
  }

  def receive = { case _ => }
}

/*
To launch it: java akka.Main Main
It outputs:

starting
started
printing List(3, 5, 1, 2)
printing 4
printing 6
printing 8
printing 9
printing 7
[INFO] [06/21/2014 23:48:27.400] [Main-akka.actor.default-dispatcher-2] [akka://Main/user/app-terminator] application supervisor has terminated, shutting down
Disconnected from the target VM, address: '127.0.0.1:57405', transport: 'socket'

Process finished with exit code 0











*/