Created
May 21, 2012 20:18
-
-
Save derekjw/2764401 to your computer and use it in GitHub Desktop.
Safely closing over actor state
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import akka.actor.Actor | |
object RequestCont { | |
def apply[A](actor: Actor)(cont: A => Unit): RequestCont[A] = new RequestCont[A](actor, cont) | |
} | |
class RequestCont[A](owner: Actor, cont: A => Unit) extends (A => ResponseCont) { | |
def apply(value: A): ResponseCont = { | |
ResponseCont(owner)(cont(value)) | |
} | |
} |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import akka.actor.Actor | |
object ResponseCont { | |
def apply(owner: Actor)(cont: => Unit): ResponseCont = new ResponseCont(owner, cont) | |
} | |
class ResponseCont(owner: Actor, cont: => Unit) extends (Actor => Boolean) { | |
def apply(actor: Actor): Boolean = { | |
if (owner eq actor) { | |
cont | |
true | |
} else false | |
} | |
} |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import akka.actor.{Props, ActorSystem, ActorRef, Actor} | |
import akka.pattern.ask | |
import akka.dispatch.{Await, Future} | |
import akka.util.{Timeout, Duration} | |
import akka.util.duration._ | |
import org.junit.{After, Test} | |
import org.junit.Assert._ | |
case object Incr | |
case class Request(n: Int, cont: RequestCont[Int]) | |
class Requester(responder: ActorRef) extends Actor { | |
var counter = 0 | |
def receive = { | |
case cont: ResponseCont => cont(this) | |
case Incr => | |
val sender = context.sender | |
responder ! Request(1, RequestCont(this) { (n: Int) => | |
counter += n | |
sender ! counter | |
}) | |
} | |
} | |
class Responder extends Actor { | |
def receive = { | |
case Request(n, cont) => | |
sender ! cont(n + 1) | |
} | |
} | |
class ResponseContTest { | |
implicit val system = ActorSystem() | |
implicit val timeout = Timeout(5 seconds) | |
val responder = system actorOf Props[Responder] | |
val requester = system actorOf (Props(new Requester(responder))) | |
@After | |
def tearDown() { | |
system.shutdown() | |
} | |
@Test | |
def shouldBeConsistent() { | |
val expected = List.range(1, 100) map (_ * 2) | |
val future = Future.traverse(expected)(_ => requester ? Incr) | |
assertEquals(expected, Await result (future, Duration.Inf)) | |
} | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment