Merge branch 'master' into wip-2537-doc-remote-deploy-java-patriknw

Conflicts:
	akka-docs/rst/java/code/docs/remoting/RemoteActorExample.java
This commit is contained in:
Patrik Nordwall 2012-09-21 17:01:13 +02:00
commit 5a2ebb172d
273 changed files with 356 additions and 205 deletions

View file

@ -0,0 +1,409 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import language.postfixOps
//#imports1
import akka.actor.Actor
import akka.actor.Props
import akka.event.Logging
//#imports1
import scala.concurrent.Future
import akka.actor.{ ActorRef, ActorSystem }
import org.scalatest.{ BeforeAndAfterAll, WordSpec }
import org.scalatest.matchers.MustMatchers
import akka.testkit._
import akka.util._
import scala.concurrent.util.duration._
import akka.actor.Actor.Receive
import scala.concurrent.Await
//#my-actor
class MyActor extends Actor {
val log = Logging(context.system, this)
def receive = {
case "test" log.info("received test")
case _ log.info("received unknown message")
}
}
//#my-actor
case class DoIt(msg: ImmutableMessage)
case class Message(s: String)
//#context-actorOf
class FirstActor extends Actor {
val myActor = context.actorOf(Props[MyActor], name = "myactor")
//#context-actorOf
def receive = {
case x sender ! x
}
}
class AnonymousActor extends Actor {
//#anonymous-actor
def receive = {
case m: DoIt
context.actorOf(Props(new Actor {
def receive = {
case DoIt(msg)
val replyMsg = doSomeDangerousWork(msg)
sender ! replyMsg
context.stop(self)
}
def doSomeDangerousWork(msg: ImmutableMessage): String = { "done" }
})) forward m
}
//#anonymous-actor
}
//#system-actorOf
object Main extends App {
val system = ActorSystem("MySystem")
val myActor = system.actorOf(Props[MyActor], name = "myactor")
//#system-actorOf
}
class ReplyException extends Actor {
def receive = {
case _
//#reply-exception
try {
val result = operation()
sender ! result
} catch {
case e: Exception
sender ! akka.actor.Status.Failure(e)
throw e
}
//#reply-exception
}
def operation(): String = { "Hi" }
}
//#swapper
case object Swap
class Swapper extends Actor {
import context._
val log = Logging(system, this)
def receive = {
case Swap
log.info("Hi")
become {
case Swap
log.info("Ho")
unbecome() // resets the latest 'become' (just for fun)
}
}
}
object SwapperApp extends App {
val system = ActorSystem("SwapperSystem")
val swap = system.actorOf(Props[Swapper], name = "swapper")
swap ! Swap // logs Hi
swap ! Swap // logs Ho
swap ! Swap // logs Hi
swap ! Swap // logs Ho
swap ! Swap // logs Hi
swap ! Swap // logs Ho
}
//#swapper
//#receive-orElse
abstract class GenericActor extends Actor {
// to be defined in subclassing actor
def specificMessageHandler: Receive
// generic message handler
def genericMessageHandler: Receive = {
case event printf("generic: %s\n", event)
}
def receive = specificMessageHandler orElse genericMessageHandler
}
class SpecificActor extends GenericActor {
def specificMessageHandler = {
case event: MyMsg printf("specific: %s\n", event.subject)
}
}
case class MyMsg(subject: String)
//#receive-orElse
//#receive-orElse2
trait ComposableActor extends Actor {
private var receives: List[Receive] = List()
protected def registerReceive(receive: Receive) {
receives = receive :: receives
}
def receive = receives reduce { _ orElse _ }
}
class MyComposableActor extends ComposableActor {
override def preStart() {
registerReceive({
case "foo" /* Do something */
})
registerReceive({
case "bar" /* Do something */
})
}
}
//#receive-orElse2
class ActorDocSpec extends AkkaSpec(Map("akka.loglevel" -> "INFO")) {
"import context" in {
//#import-context
class FirstActor extends Actor {
import context._
val myActor = actorOf(Props[MyActor], name = "myactor")
def receive = {
case x myActor ! x
}
}
//#import-context
val first = system.actorOf(Props(new FirstActor), name = "first")
system.stop(first)
}
"creating actor with AkkaSpec.actorOf" in {
val myActor = system.actorOf(Props[MyActor])
// testing the actor
// TODO: convert docs to AkkaSpec(Map(...))
val filter = EventFilter.custom {
case e: Logging.Info true
case _ false
}
system.eventStream.publish(TestEvent.Mute(filter))
system.eventStream.subscribe(testActor, classOf[Logging.Info])
myActor ! "test"
expectMsgPF(1 second) { case Logging.Info(_, _, "received test") true }
myActor ! "unknown"
expectMsgPF(1 second) { case Logging.Info(_, _, "received unknown message") true }
system.eventStream.unsubscribe(testActor)
system.eventStream.publish(TestEvent.UnMute(filter))
system.stop(myActor)
}
"creating actor with constructor" in {
class MyActor(arg: String) extends Actor {
def receive = { case _ () }
}
//#creating-constructor
// allows passing in arguments to the MyActor constructor
val myActor = system.actorOf(Props(new MyActor("...")), name = "myactor")
//#creating-constructor
system.stop(myActor)
}
"creating a Props config" in {
//#creating-props-config
import akka.actor.Props
val props1 = Props.empty
val props2 = Props[MyActor]
val props3 = Props(new MyActor)
val props4 = Props(
creator = { () new MyActor },
dispatcher = "my-dispatcher")
val props5 = props1.withCreator(new MyActor)
val props6 = props5.withDispatcher("my-dispatcher")
//#creating-props-config
}
"creating actor with Props" in {
//#creating-props
import akka.actor.Props
val myActor = system.actorOf(Props[MyActor].withDispatcher("my-dispatcher"), name = "myactor2")
//#creating-props
system.stop(myActor)
}
"using implicit timeout" in {
val myActor = system.actorOf(Props(new FirstActor))
//#using-implicit-timeout
import scala.concurrent.util.duration._
import akka.util.Timeout
import akka.pattern.ask
implicit val timeout = Timeout(5 seconds)
val future = myActor ? "hello"
//#using-implicit-timeout
Await.result(future, timeout.duration) must be("hello")
}
"using explicit timeout" in {
val myActor = system.actorOf(Props(new FirstActor))
//#using-explicit-timeout
import scala.concurrent.util.duration._
import akka.pattern.ask
val future = myActor.ask("hello")(5 seconds)
//#using-explicit-timeout
Await.result(future, 5 seconds) must be("hello")
}
"using receiveTimeout" in {
//#receive-timeout
import akka.actor.ReceiveTimeout
import scala.concurrent.util.duration._
class MyActor extends Actor {
context.setReceiveTimeout(30 milliseconds)
def receive = {
case "Hello" //...
case ReceiveTimeout throw new RuntimeException("received timeout")
}
}
//#receive-timeout
}
"using hot-swap" in {
//#hot-swap-actor
class HotSwapActor extends Actor {
import context._
def angry: Receive = {
case "foo" sender ! "I am already angry?"
case "bar" become(happy)
}
def happy: Receive = {
case "bar" sender ! "I am already happy :-)"
case "foo" become(angry)
}
def receive = {
case "foo" become(angry)
case "bar" become(happy)
}
}
//#hot-swap-actor
val actor = system.actorOf(Props(new HotSwapActor), name = "hot")
}
"using Stash" in {
//#stash
import akka.actor.Stash
class ActorWithProtocol extends Actor with Stash {
def receive = {
case "open"
unstashAll()
context.become {
case "write" // do writing...
case "close"
unstashAll()
context.unbecome()
case msg stash()
}
case msg stash()
}
}
//#stash
}
"using watch" in {
//#watch
import akka.actor.{ Actor, Props, Terminated }
class WatchActor extends Actor {
val child = context.actorOf(Props.empty, "child")
context.watch(child) // <-- this is the only call needed for registration
var lastSender = system.deadLetters
def receive = {
case "kill" context.stop(child); lastSender = sender
case Terminated(`child`) lastSender ! "finished"
}
}
//#watch
val a = system.actorOf(Props(new WatchActor))
implicit val sender = testActor
a ! "kill"
expectMsg("finished")
}
"using pattern gracefulStop" in {
val actorRef = system.actorOf(Props[MyActor])
//#gracefulStop
import akka.pattern.gracefulStop
import scala.concurrent.Await
try {
val stopped: Future[Boolean] = gracefulStop(actorRef, 5 seconds)(system)
Await.result(stopped, 6 seconds)
// the actor has been stopped
} catch {
case e: akka.pattern.AskTimeoutException // the actor wasn't stopped within 5 seconds
}
//#gracefulStop
}
"using pattern ask / pipeTo" in {
val actorA, actorB, actorC, actorD = system.actorOf(Props.empty)
//#ask-pipeTo
import akka.pattern.{ ask, pipe }
import system.dispatcher // The ExecutionContext that will be used
case class Result(x: Int, s: String, d: Double)
case object Request
implicit val timeout = Timeout(5 seconds) // needed for `?` below
val f: Future[Result] =
for {
x ask(actorA, Request).mapTo[Int] // call pattern directly
s (actorB ask Request).mapTo[String] // call by implicit conversion
d (actorC ? Request).mapTo[Double] // call by symbolic name
} yield Result(x, s, d)
f pipeTo actorD // .. or ..
pipe(f) to actorD
//#ask-pipeTo
}
"replying with own or other sender" in {
val actor = system.actorOf(Props(new Actor {
def receive = {
case ref: ActorRef
//#reply-with-sender
sender.tell("reply", context.parent) // replies will go back to parent
sender.!("reply")(context.parent) // alternative syntax (beware of the parens!)
//#reply-with-sender
case x
//#reply-without-sender
sender ! x // replies will go to this actor
//#reply-without-sender
}
}))
implicit val me = testActor
actor ! 42
expectMsg(42)
lastSender must be === actor
actor ! me
expectMsg("reply")
lastSender must be === system.actorFor("/user")
expectMsg("reply")
lastSender must be === system.actorFor("/user")
}
}

View file

@ -0,0 +1,211 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import language.postfixOps
import akka.testkit.{ AkkaSpec MyFavoriteTestFrameWorkPlusAkkaTestKit }
//#test-code
import akka.actor.Props
class FSMDocSpec extends MyFavoriteTestFrameWorkPlusAkkaTestKit {
"simple finite state machine" must {
//#fsm-code-elided
//#simple-imports
import akka.actor.{ Actor, ActorRef, FSM }
import scala.concurrent.util.duration._
//#simple-imports
//#simple-events
// received events
case class SetTarget(ref: ActorRef)
case class Queue(obj: Any)
case object Flush
// sent events
case class Batch(obj: Seq[Any])
//#simple-events
//#simple-state
// states
sealed trait State
case object Idle extends State
case object Active extends State
sealed trait Data
case object Uninitialized extends Data
case class Todo(target: ActorRef, queue: Seq[Any]) extends Data
//#simple-state
//#simple-fsm
class Buncher extends Actor with FSM[State, Data] {
//#fsm-body
startWith(Idle, Uninitialized)
//#when-syntax
when(Idle) {
case Event(SetTarget(ref), Uninitialized)
stay using Todo(ref, Vector.empty)
}
//#when-syntax
//#transition-elided
onTransition {
case Active -> Idle
stateData match {
case Todo(ref, queue) ref ! Batch(queue)
}
}
//#transition-elided
//#when-syntax
when(Active, stateTimeout = 1 second) {
case Event(Flush | StateTimeout, t: Todo)
goto(Idle) using t.copy(queue = Vector.empty)
}
//#when-syntax
//#unhandled-elided
whenUnhandled {
// common code for both states
case Event(Queue(obj), t @ Todo(_, v))
goto(Active) using t.copy(queue = v :+ obj)
case Event(e, s)
log.warning("received unhandled request {} in state {}/{}", e, stateName, s)
stay
}
//#unhandled-elided
//#fsm-body
initialize
}
//#simple-fsm
object DemoCode {
trait StateType
case object SomeState extends StateType
case object Processing extends StateType
case object Error extends StateType
case object Idle extends StateType
case object Active extends StateType
class Dummy extends Actor with FSM[StateType, Int] {
class X
val newData = 42
object WillDo
object Tick
//#modifier-syntax
when(SomeState) {
case Event(msg, _)
goto(Processing) using (newData) forMax (5 seconds) replying (WillDo)
}
//#modifier-syntax
//#transition-syntax
onTransition {
case Idle -> Active setTimer("timeout", Tick, 1 second, true)
case Active -> _ cancelTimer("timeout")
case x -> Idle log.info("entering Idle from " + x)
}
//#transition-syntax
//#alt-transition-syntax
onTransition(handler _)
def handler(from: StateType, to: StateType) {
// handle it here ...
}
//#alt-transition-syntax
//#stop-syntax
when(Error) {
case Event("stop", _)
// do cleanup ...
stop()
}
//#stop-syntax
//#transform-syntax
when(SomeState)(transform {
case Event(bytes: Array[Byte], read) stay using (read + bytes.length)
case Event(bytes: List[Byte], read) stay using (read + bytes.size)
} using {
case s @ FSM.State(state, read, timeout, stopReason, replies) if read > 1000
goto(Processing)
})
//#transform-syntax
//#alt-transform-syntax
val processingTrigger: PartialFunction[State, State] = {
case s @ FSM.State(state, read, timeout, stopReason, replies) if read > 1000
goto(Processing)
}
when(SomeState)(transform {
case Event(bytes: Array[Byte], read) stay using (read + bytes.length)
case Event(bytes: List[Byte], read) stay using (read + bytes.size)
} using processingTrigger)
//#alt-transform-syntax
//#termination-syntax
onTermination {
case StopEvent(FSM.Normal, state, data) // ...
case StopEvent(FSM.Shutdown, state, data) // ...
case StopEvent(FSM.Failure(cause), state, data) // ...
}
//#termination-syntax
//#unhandled-syntax
whenUnhandled {
case Event(x: X, data)
log.info("Received unhandled event: " + x)
stay
case Event(msg, _)
log.warning("Received unknown event: " + msg)
goto(Error)
}
//#unhandled-syntax
}
//#logging-fsm
import akka.actor.LoggingFSM
class MyFSM extends Actor with LoggingFSM[StateType, Data] {
//#body-elided
override def logDepth = 12
onTermination {
case StopEvent(FSM.Failure(_), state, data)
val lastEvents = getLog.mkString("\n\t")
log.warning("Failure in state " + state + " with data " + data + "\n" +
"Events leading up to this point:\n\t" + lastEvents)
}
// ...
//#body-elided
}
//#logging-fsm
}
//#fsm-code-elided
"batch correctly" in {
val buncher = system.actorOf(Props(new Buncher))
buncher ! SetTarget(testActor)
buncher ! Queue(42)
buncher ! Queue(43)
expectMsg(Batch(Seq(42, 43)))
buncher ! Queue(44)
buncher ! Flush
buncher ! Queue(45)
expectMsg(Batch(Seq(44)))
expectMsg(Batch(Seq(45)))
}
"batch not if uninitialized" in {
val buncher = system.actorOf(Props(new Buncher))
buncher ! Queue(42)
expectNoMsg
}
}
}
//#test-code

View file

@ -0,0 +1,294 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import language.postfixOps
//#all
//#imports
import akka.actor._
import akka.actor.SupervisorStrategy._
import scala.concurrent.util.duration._
import scala.concurrent.util.Duration
import akka.util.Timeout
import akka.event.LoggingReceive
import akka.pattern.{ ask, pipe }
import com.typesafe.config.ConfigFactory
//#imports
/**
* Runs the sample
*/
object FaultHandlingDocSample extends App {
import Worker._
val config = ConfigFactory.parseString("""
akka.loglevel = DEBUG
akka.actor.debug {
receive = on
lifecycle = on
}
""")
val system = ActorSystem("FaultToleranceSample", config)
val worker = system.actorOf(Props[Worker], name = "worker")
val listener = system.actorOf(Props[Listener], name = "listener")
// start the work and listen on progress
// note that the listener is used as sender of the tell,
// i.e. it will receive replies from the worker
worker.tell(Start, sender = listener)
}
/**
* Listens on progress from the worker and shuts down the system when enough
* work has been done.
*/
class Listener extends Actor with ActorLogging {
import Worker._
// If we don't get any progress within 15 seconds then the service is unavailable
context.setReceiveTimeout(15 seconds)
def receive = {
case Progress(percent)
log.info("Current progress: {} %", percent)
if (percent >= 100.0) {
log.info("That's all, shutting down")
context.system.shutdown()
}
case ReceiveTimeout
// No progress within 15 seconds, ServiceUnavailable
log.error("Shutting down due to unavailable service")
context.system.shutdown()
}
}
//#messages
object Worker {
case object Start
case object Do
case class Progress(percent: Double)
}
//#messages
/**
* Worker performs some work when it receives the `Start` message.
* It will continuously notify the sender of the `Start` message
* of current ``Progress``. The `Worker` supervise the `CounterService`.
*/
class Worker extends Actor with ActorLogging {
import Worker._
import CounterService._
implicit val askTimeout = Timeout(5 seconds)
// Stop the CounterService child if it throws ServiceUnavailable
override val supervisorStrategy = OneForOneStrategy() {
case _: CounterService.ServiceUnavailable Stop
}
// The sender of the initial Start message will continuously be notified about progress
var progressListener: Option[ActorRef] = None
val counterService = context.actorOf(Props[CounterService], name = "counter")
val totalCount = 51
import context.dispatcher // Use this Actors' Dispatcher as ExecutionContext
def receive = LoggingReceive {
case Start if progressListener.isEmpty
progressListener = Some(sender)
context.system.scheduler.schedule(Duration.Zero, 1 second, self, Do)
case Do
counterService ! Increment(1)
counterService ! Increment(1)
counterService ! Increment(1)
// Send current progress to the initial sender
counterService ? GetCurrentCount map {
case CurrentCount(_, count) Progress(100.0 * count / totalCount)
} pipeTo progressListener.get
}
}
//#messages
object CounterService {
case class Increment(n: Int)
case object GetCurrentCount
case class CurrentCount(key: String, count: Long)
class ServiceUnavailable(msg: String) extends RuntimeException(msg)
private case object Reconnect
}
//#messages
/**
* Adds the value received in `Increment` message to a persistent
* counter. Replies with `CurrentCount` when it is asked for `CurrentCount`.
* `CounterService` supervise `Storage` and `Counter`.
*/
class CounterService extends Actor {
import CounterService._
import Counter._
import Storage._
// Restart the storage child when StorageException is thrown.
// After 3 restarts within 5 seconds it will be stopped.
override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 3, withinTimeRange = 5 seconds) {
case _: Storage.StorageException Restart
}
val key = self.path.name
var storage: Option[ActorRef] = None
var counter: Option[ActorRef] = None
var backlog = IndexedSeq.empty[(ActorRef, Any)]
val MaxBacklog = 10000
import context.dispatcher // Use this Actors' Dispatcher as ExecutionContext
override def preStart() {
initStorage()
}
/**
* The child storage is restarted in case of failure, but after 3 restarts,
* and still failing it will be stopped. Better to back-off than continuously
* failing. When it has been stopped we will schedule a Reconnect after a delay.
* Watch the child so we receive Terminated message when it has been terminated.
*/
def initStorage() {
storage = Some(context.watch(context.actorOf(Props[Storage], name = "storage")))
// Tell the counter, if any, to use the new storage
counter foreach { _ ! UseStorage(storage) }
// We need the initial value to be able to operate
storage.get ! Get(key)
}
def receive = LoggingReceive {
case Entry(k, v) if k == key && counter == None
// Reply from Storage of the initial value, now we can create the Counter
val c = context.actorOf(Props(new Counter(key, v)))
counter = Some(c)
// Tell the counter to use current storage
c ! UseStorage(storage)
// and send the buffered backlog to the counter
for ((replyTo, msg) backlog) c.tell(msg, sender = replyTo)
backlog = IndexedSeq.empty
case msg @ Increment(n) forwardOrPlaceInBacklog(msg)
case msg @ GetCurrentCount forwardOrPlaceInBacklog(msg)
case Terminated(actorRef) if Some(actorRef) == storage
// After 3 restarts the storage child is stopped.
// We receive Terminated because we watch the child, see initStorage.
storage = None
// Tell the counter that there is no storage for the moment
counter foreach { _ ! UseStorage(None) }
// Try to re-establish storage after while
context.system.scheduler.scheduleOnce(10 seconds, self, Reconnect)
case Reconnect
// Re-establish storage after the scheduled delay
initStorage()
}
def forwardOrPlaceInBacklog(msg: Any) {
// We need the initial value from storage before we can start delegate to the counter.
// Before that we place the messages in a backlog, to be sent to the counter when
// it is initialized.
counter match {
case Some(c) c forward msg
case None
if (backlog.size >= MaxBacklog)
throw new ServiceUnavailable("CounterService not available, lack of initial value")
backlog = backlog :+ (sender, msg)
}
}
}
//#messages
object Counter {
case class UseStorage(storage: Option[ActorRef])
}
//#messages
/**
* The in memory count variable that will send current
* value to the `Storage`, if there is any storage
* available at the moment.
*/
class Counter(key: String, initialValue: Long) extends Actor {
import Counter._
import CounterService._
import Storage._
var count = initialValue
var storage: Option[ActorRef] = None
def receive = LoggingReceive {
case UseStorage(s)
storage = s
storeCount()
case Increment(n)
count += n
storeCount()
case GetCurrentCount
sender ! CurrentCount(key, count)
}
def storeCount() {
// Delegate dangerous work, to protect our valuable state.
// We can continue without storage.
storage foreach { _ ! Store(Entry(key, count)) }
}
}
//#messages
object Storage {
case class Store(entry: Entry)
case class Get(key: String)
case class Entry(key: String, value: Long)
class StorageException(msg: String) extends RuntimeException(msg)
}
//#messages
/**
* Saves key/value pairs to persistent storage when receiving `Store` message.
* Replies with current value when receiving `Get` message.
* Will throw StorageException if the underlying data store is out of order.
*/
class Storage extends Actor {
import Storage._
val db = DummyDB
def receive = LoggingReceive {
case Store(Entry(key, count)) db.save(key, count)
case Get(key) sender ! Entry(key, db.load(key).getOrElse(0L))
}
}
//#dummydb
object DummyDB {
import Storage.StorageException
private var db = Map[String, Long]()
@throws(classOf[StorageException])
def save(key: String, value: Long): Unit = synchronized {
if (11 <= value && value <= 14) throw new StorageException("Simulated store failure " + value)
db += (key -> value)
}
@throws(classOf[StorageException])
def load(key: String): Option[Long] = synchronized {
db.get(key)
}
}
//#dummydb
//#all

View file

@ -0,0 +1,156 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import language.postfixOps
//#testkit
import akka.testkit.{ AkkaSpec, ImplicitSender, EventFilter }
import akka.actor.{ ActorRef, Props, Terminated }
//#testkit
object FaultHandlingDocSpec {
//#supervisor
//#child
import akka.actor.Actor
//#child
//#supervisor
//#supervisor
class Supervisor extends Actor {
//#strategy
import akka.actor.OneForOneStrategy
import akka.actor.SupervisorStrategy._
import scala.concurrent.util.duration._
override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
case _: ArithmeticException Resume
case _: NullPointerException Restart
case _: IllegalArgumentException Stop
case _: Exception Escalate
}
//#strategy
def receive = {
case p: Props sender ! context.actorOf(p)
}
}
//#supervisor
//#supervisor2
class Supervisor2 extends Actor {
//#strategy2
import akka.actor.OneForOneStrategy
import akka.actor.SupervisorStrategy._
import scala.concurrent.util.duration._
override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
case _: ArithmeticException Resume
case _: NullPointerException Restart
case _: IllegalArgumentException Stop
case _: Exception Escalate
}
//#strategy2
def receive = {
case p: Props sender ! context.actorOf(p)
}
// override default to kill all children during restart
override def preRestart(cause: Throwable, msg: Option[Any]) {}
}
//#supervisor2
//#child
class Child extends Actor {
var state = 0
def receive = {
case ex: Exception throw ex
case x: Int state = x
case "get" sender ! state
}
}
//#child
}
//#testkit
class FaultHandlingDocSpec extends AkkaSpec with ImplicitSender {
//#testkit
import FaultHandlingDocSpec._
//#testkit
"A supervisor" must {
"apply the chosen strategy for its child" in {
//#testkit
//#create
val supervisor = system.actorOf(Props[Supervisor], "supervisor")
supervisor ! Props[Child]
val child = expectMsgType[ActorRef] // retrieve answer from TestKits testActor
//#create
EventFilter[ArithmeticException](occurrences = 1) intercept {
//#resume
child ! 42 // set state to 42
child ! "get"
expectMsg(42)
child ! new ArithmeticException // crash it
child ! "get"
expectMsg(42)
//#resume
}
EventFilter[NullPointerException](occurrences = 1) intercept {
//#restart
child ! new NullPointerException // crash it harder
child ! "get"
expectMsg(0)
//#restart
}
EventFilter[IllegalArgumentException](occurrences = 1) intercept {
//#stop
watch(child) // have testActor watch child
child ! new IllegalArgumentException // break it
expectMsgPF() {
case t @ Terminated(`child`) if t.existenceConfirmed ()
}
child.isTerminated must be(true)
//#stop
}
EventFilter[Exception]("CRASH", occurrences = 4) intercept {
//#escalate-kill
supervisor ! Props[Child] // create new child
val child2 = expectMsgType[ActorRef]
watch(child2)
child2 ! "get" // verify it is alive
expectMsg(0)
child2 ! new Exception("CRASH") // escalate failure
expectMsgPF() {
case t @ Terminated(`child2`) if t.existenceConfirmed ()
}
//#escalate-kill
//#escalate-restart
val supervisor2 = system.actorOf(Props[Supervisor2], "supervisor2")
supervisor2 ! Props[Child]
val child3 = expectMsgType[ActorRef]
child3 ! 23
child3 ! "get"
expectMsg(23)
child3 ! new Exception("CRASH")
child3 ! "get"
expectMsg(0)
//#escalate-restart
}
//#testkit
// code here
}
}
}
//#testkit

View file

@ -0,0 +1,64 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import language.postfixOps
//#imports1
import akka.actor.Actor
import akka.actor.Props
import scala.concurrent.util.duration._
//#imports1
import org.scalatest.{ BeforeAndAfterAll, WordSpec }
import org.scalatest.matchers.MustMatchers
import akka.testkit._
class SchedulerDocSpec extends AkkaSpec(Map("akka.loglevel" -> "INFO")) {
"schedule a one-off task" in {
//#schedule-one-off-message
//Use the system's dispatcher as ExecutionContext
import system.dispatcher
//Schedules to send the "foo"-message to the testActor after 50ms
system.scheduler.scheduleOnce(50 milliseconds, testActor, "foo")
//#schedule-one-off-message
expectMsg(1 second, "foo")
//#schedule-one-off-thunk
//Schedules a function to be executed (send the current time) to the testActor after 50ms
system.scheduler.scheduleOnce(50 milliseconds) {
testActor ! System.currentTimeMillis
}
//#schedule-one-off-thunk
}
"schedule a recurring task" in {
//#schedule-recurring
val Tick = "tick"
val tickActor = system.actorOf(Props(new Actor {
def receive = {
case Tick //Do something
}
}))
//Use system's dispatcher as ExecutionContext
import system.dispatcher
//This will schedule to send the Tick-message
//to the tickActor after 0ms repeating every 50ms
val cancellable =
system.scheduler.schedule(0 milliseconds,
50 milliseconds,
tickActor,
Tick)
//This cancels further Ticks to be sent
cancellable.cancel()
//#schedule-recurring
system.stop(tickActor)
}
}

View file

@ -0,0 +1,179 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import language.postfixOps
//#imports
import scala.concurrent.{ Promise, Future, Await }
import scala.concurrent.util.duration._
import akka.actor.{ ActorContext, TypedActor, TypedProps }
//#imports
import org.scalatest.{ BeforeAndAfterAll, WordSpec }
import org.scalatest.matchers.MustMatchers
import akka.testkit._
//#typed-actor-iface
trait Squarer {
//#typed-actor-iface-methods
def squareDontCare(i: Int): Unit //fire-forget
def square(i: Int): Future[Int] //non-blocking send-request-reply
def squareNowPlease(i: Int): Option[Int] //blocking send-request-reply
def squareNow(i: Int): Int //blocking send-request-reply
//#typed-actor-iface-methods
}
//#typed-actor-iface
//#typed-actor-impl
class SquarerImpl(val name: String) extends Squarer {
def this() = this("default")
//#typed-actor-impl-methods
import TypedActor.dispatcher //So we can create Promises
def squareDontCare(i: Int): Unit = i * i //Nobody cares :(
def square(i: Int): Future[Int] = Promise.successful(i * i).future
def squareNowPlease(i: Int): Option[Int] = Some(i * i)
def squareNow(i: Int): Int = i * i
//#typed-actor-impl-methods
}
//#typed-actor-impl
import java.lang.String.{ valueOf println } //Mr funny man avoids printing to stdout AND keeping docs alright
//#typed-actor-supercharge
trait Foo {
def doFoo(times: Int): Unit = println("doFoo(" + times + ")")
}
trait Bar {
import TypedActor.dispatcher //So we have an implicit dispatcher for our Promise
def doBar(str: String): Future[String] = Promise.successful(str.toUpperCase).future
}
class FooBar extends Foo with Bar
//#typed-actor-supercharge
class TypedActorDocSpec extends AkkaSpec(Map("akka.loglevel" -> "INFO")) {
"get the TypedActor extension" in {
val someReference: AnyRef = null
try {
//#typed-actor-extension-tools
import akka.actor.TypedActor
//Returns the Typed Actor Extension
val extension = TypedActor(system) //system is an instance of ActorSystem
//Returns whether the reference is a Typed Actor Proxy or not
TypedActor(system).isTypedActor(someReference)
//Returns the backing Akka Actor behind an external Typed Actor Proxy
TypedActor(system).getActorRefFor(someReference)
//Returns the current ActorContext,
// method only valid within methods of a TypedActor implementation
val c: ActorContext = TypedActor.context
//Returns the external proxy of the current Typed Actor,
// method only valid within methods of a TypedActor implementation
val s: Squarer = TypedActor.self[Squarer]
//Returns a contextual instance of the Typed Actor Extension
//this means that if you create other Typed Actors with this,
//they will become children to the current Typed Actor.
TypedActor(TypedActor.context)
//#typed-actor-extension-tools
} catch {
case e: Exception //dun care
}
}
"create a typed actor" in {
//#typed-actor-create1
val mySquarer: Squarer =
TypedActor(system).typedActorOf(TypedProps[SquarerImpl]())
//#typed-actor-create1
//#typed-actor-create2
val otherSquarer: Squarer =
TypedActor(system).typedActorOf(TypedProps(classOf[Squarer], new SquarerImpl("foo")), "name")
//#typed-actor-create2
//#typed-actor-calls
//#typed-actor-call-oneway
mySquarer.squareDontCare(10)
//#typed-actor-call-oneway
//#typed-actor-call-future
val fSquare = mySquarer.square(10) //A Future[Int]
//#typed-actor-call-future
//#typed-actor-call-option
val oSquare = mySquarer.squareNowPlease(10) //Option[Int]
//#typed-actor-call-option
//#typed-actor-call-strict
val iSquare = mySquarer.squareNow(10) //Int
//#typed-actor-call-strict
//#typed-actor-calls
Await.result(fSquare, 3 seconds) must be === 100
oSquare must be === Some(100)
iSquare must be === 100
//#typed-actor-stop
TypedActor(system).stop(mySquarer)
//#typed-actor-stop
//#typed-actor-poisonpill
TypedActor(system).poisonPill(otherSquarer)
//#typed-actor-poisonpill
}
"proxy any ActorRef" in {
//#typed-actor-remote
val typedActor: Foo with Bar =
TypedActor(system).
typedActorOf(
TypedProps[FooBar],
system.actorFor("akka://SomeSystem@somehost:2552/user/some/foobar"))
//Use "typedActor" as a FooBar
//#typed-actor-remote
}
"create hierarchies" in {
try {
//#typed-actor-hierarchy
//Inside your Typed Actor
val childSquarer: Squarer = TypedActor(TypedActor.context).typedActorOf(TypedProps[SquarerImpl]())
//Use "childSquarer" as a Squarer
//#typed-actor-hierarchy
} catch {
case e: Exception //ignore
}
}
"supercharge" in {
//#typed-actor-supercharge-usage
val awesomeFooBar: Foo with Bar = TypedActor(system).typedActorOf(TypedProps[FooBar]())
awesomeFooBar.doFoo(10)
val f = awesomeFooBar.doBar("yes")
TypedActor(system).poisonPill(awesomeFooBar)
//#typed-actor-supercharge-usage
Await.result(f, 3 seconds) must be === "YES"
}
}

View file

@ -0,0 +1,50 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.actor
import akka.actor._
import scala.collection.mutable.ListBuffer
/**
* Requirements are as follows:
* The first thing the actor needs to do, is to subscribe to a channel of events,
* Then it must replay (process) all "old" events
* Then it has to wait for a GoAhead signal to begin processing the new events
* It mustn't "miss" events that happen between catching up with the old events and getting the GoAhead signal
*/
class UnnestedReceives extends Actor {
import context.become
//If you need to store sender/senderFuture you can change it to ListBuffer[(Any, Channel)]
val queue = new ListBuffer[Any]()
//This message processes a message/event
def process(msg: Any): Unit = println("processing: " + msg)
//This method subscribes the actor to the event bus
def subscribe() {} //Your external stuff
//This method retrieves all prior messages/events
def allOldMessages() = List()
override def preStart {
//We override preStart to be sure that the first message the actor gets is
//'Replay, that message will start to be processed _after_ the actor is started
self ! 'Replay
//Then we subscribe to the stream of messages/events
subscribe()
}
def receive = {
case 'Replay //Our first message should be a 'Replay message, all others are invalid
allOldMessages() foreach process //Process all old messages/events
become { //Switch behavior to look for the GoAhead signal
case 'GoAhead //When we get the GoAhead signal we process all our buffered messages/events
queue foreach process
queue.clear
become { //Then we change behaviour to process incoming messages/events as they arrive
case msg process(msg)
}
case msg //While we haven't gotten the GoAhead signal, buffer all incoming messages
queue += msg //Here you have full control, you can handle overflow etc
}
}
}

View file

@ -0,0 +1,192 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.agent
import language.postfixOps
import akka.agent.Agent
import scala.concurrent.util.duration._
import akka.util.Timeout
import akka.testkit._
class AgentDocSpec extends AkkaSpec {
"create and close" in {
//#create
import akka.agent.Agent
val agent = Agent(5)
//#create
//#close
agent.close()
//#close
}
"create with implicit system" in {
//#create-implicit-system
import akka.actor.ActorSystem
import akka.agent.Agent
implicit val system = ActorSystem("app")
val agent = Agent(5)
//#create-implicit-system
agent.close()
system.shutdown()
}
"create with explicit system" in {
//#create-explicit-system
import akka.actor.ActorSystem
import akka.agent.Agent
val system = ActorSystem("app")
val agent = Agent(5)(system)
//#create-explicit-system
agent.close()
system.shutdown()
}
"send and sendOff" in {
val agent = Agent(0)
import system.dispatcher
//#send
// send a value
agent send 7
// send a function
agent send (_ + 1)
agent send (_ * 2)
//#send
def longRunningOrBlockingFunction = (i: Int) i * 1
//#send-off
// sendOff a function
agent sendOff (longRunningOrBlockingFunction)
//#send-off
val result = agent.await(Timeout(5 seconds))
result must be === 16
}
"read with apply" in {
val agent = Agent(0)
//#read-apply
val result = agent()
//#read-apply
result must be === 0
}
"read with get" in {
val agent = Agent(0)
//#read-get
val result = agent.get
//#read-get
result must be === 0
}
"read with await" in {
val agent = Agent(0)
//#read-await
import scala.concurrent.util.duration._
import akka.util.Timeout
implicit val timeout = Timeout(5 seconds)
val result = agent.await
//#read-await
result must be === 0
}
"read with future" in {
val agent = Agent(0)
//#read-future
import scala.concurrent.Await
implicit val timeout = Timeout(5 seconds)
val future = agent.future
val result = Await.result(future, timeout.duration)
//#read-future
result must be === 0
}
"transfer example" in {
//#transfer-example
import akka.agent.Agent
import scala.concurrent.util.duration._
import akka.util.Timeout
import scala.concurrent.stm._
def transfer(from: Agent[Int], to: Agent[Int], amount: Int): Boolean = {
atomic { txn
if (from.get < amount) false
else {
from send (_ - amount)
to send (_ + amount)
true
}
}
}
val from = Agent(100)
val to = Agent(20)
val ok = transfer(from, to, 50)
implicit val timeout = Timeout(5 seconds)
val fromValue = from.await // -> 50
val toValue = to.await // -> 70
//#transfer-example
fromValue must be === 50
toValue must be === 70
}
"monadic example" in {
//#monadic-example
val agent1 = Agent(3)
val agent2 = Agent(5)
// uses foreach
var result = 0
for (value agent1) {
result = value + 1
}
// uses map
val agent3 = for (value agent1) yield value + 1
// or using map directly
val agent4 = agent1 map (_ + 1)
// uses flatMap
val agent5 = for {
value1 agent1
value2 agent2
} yield value1 + value2
//#monadic-example
result must be === 4
agent3() must be === 4
agent4() must be === 4
agent5() must be === 8
agent1.close()
agent2.close()
agent3.close()
agent4.close()
agent5.close()
}
}

View file

@ -0,0 +1,73 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.camel
import language.postfixOps
object Consumers {
object Sample1 {
//#Consumer1
import akka.camel.{ CamelMessage, Consumer }
class Consumer1 extends Consumer {
def endpointUri = "file:data/input/actor"
def receive = {
case msg: CamelMessage println("received %s" format msg.bodyAs[String])
}
}
//#Consumer1
}
object Sample2 {
//#Consumer2
import akka.camel.{ CamelMessage, Consumer }
class Consumer2 extends Consumer {
def endpointUri = "jetty:http://localhost:8877/camel/default"
def receive = {
case msg: CamelMessage sender ! ("Hello %s" format msg.bodyAs[String])
}
}
//#Consumer2
}
object Sample3 {
//#Consumer3
import akka.camel.{ CamelMessage, Consumer }
import akka.camel.Ack
import akka.actor.Status.Failure
class Consumer3 extends Consumer {
override def autoAck = false
def endpointUri = "jms:queue:test"
def receive = {
case msg: CamelMessage
sender ! Ack
// on success
// ..
val someException = new Exception("e1")
// on failure
sender ! Failure(someException)
}
}
//#Consumer3
}
object Sample4 {
//#Consumer4
import akka.camel.{ CamelMessage, Consumer }
import scala.concurrent.util.duration._
class Consumer4 extends Consumer {
def endpointUri = "jetty:http://localhost:8877/camel/default"
override def replyTimeout = 500 millis
def receive = {
case msg: CamelMessage sender ! ("Hello %s" format msg.bodyAs[String])
}
}
//#Consumer4
}
}

View file

@ -0,0 +1,63 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.camel
import akka.camel.CamelMessage
import akka.actor.Status.Failure
import language.existentials
object CustomRoute {
object Sample1 {
//#CustomRoute
import akka.actor.{ Props, ActorSystem, Actor, ActorRef }
import akka.camel.{ CamelMessage, CamelExtension }
import org.apache.camel.builder.RouteBuilder
import akka.camel._
class Responder extends Actor {
def receive = {
case msg: CamelMessage
sender ! (msg.mapBody {
body: String "received %s" format body
})
}
}
class CustomRouteBuilder(system: ActorSystem, responder: ActorRef) extends RouteBuilder {
def configure {
from("jetty:http://localhost:8877/camel/custom").to(responder)
}
}
val system = ActorSystem("some-system")
val camel = CamelExtension(system)
val responder = system.actorOf(Props[Responder], name = "TestResponder")
camel.context.addRoutes(new CustomRouteBuilder(system, responder))
//#CustomRoute
}
object Sample2 {
//#ErrorThrowingConsumer
import akka.camel.Consumer
import org.apache.camel.builder.Builder
import org.apache.camel.model.RouteDefinition
class ErrorThrowingConsumer(override val endpointUri: String) extends Consumer {
def receive = {
case msg: CamelMessage throw new Exception("error: %s" format msg.body)
}
override def onRouteDefinition(rd: RouteDefinition) = {
// Catch any exception and handle it by returning the exception message as response
rd.onException(classOf[Exception]).handled(true).transform(Builder.exceptionMessage).end
}
final override def preRestart(reason: Throwable, message: Option[Any]) {
sender ! Failure(reason)
}
}
//#ErrorThrowingConsumer
}
}

View file

@ -0,0 +1,50 @@
package docs.camel
object CustomRouteExample {
{
//#CustomRouteExample
import akka.actor.{ Actor, ActorRef, Props, ActorSystem }
import akka.camel.{ CamelMessage, Consumer, Producer, CamelExtension }
import org.apache.camel.builder.RouteBuilder
import org.apache.camel.{ Exchange, Processor }
class Consumer3(transformer: ActorRef) extends Actor with Consumer {
def endpointUri = "jetty:http://0.0.0.0:8877/camel/welcome"
def receive = {
// Forward a string representation of the message body to transformer
case msg: CamelMessage transformer.forward(msg.bodyAs[String])
}
}
class Transformer(producer: ActorRef) extends Actor {
def receive = {
// example: transform message body "foo" to "- foo -" and forward result to producer
case msg: CamelMessage producer.forward(msg.mapBody((body: String) "- %s -" format body))
}
}
class Producer1 extends Actor with Producer {
def endpointUri = "direct:welcome"
}
class CustomRouteBuilder extends RouteBuilder {
def configure {
from("direct:welcome").process(new Processor() {
def process(exchange: Exchange) {
// Create a 'welcome' message from the input message
exchange.getOut.setBody("Welcome %s" format exchange.getIn.getBody)
}
})
}
}
// the below lines can be added to a Boot class, so that you can run the example from a MicroKernel
val system = ActorSystem("some-system")
val producer = system.actorOf(Props[Producer1])
val mediator = system.actorOf(Props(new Transformer(producer)))
val consumer = system.actorOf(Props(new Consumer3(mediator)))
CamelExtension(system).context.addRoutes(new CustomRouteBuilder)
//#CustomRouteExample
}
}

View file

@ -0,0 +1,47 @@
package docs.camel
object HttpExample {
{
//#HttpExample
import org.apache.camel.Exchange
import akka.actor.{ Actor, ActorRef, Props, ActorSystem }
import akka.camel.{ Producer, CamelMessage, Consumer }
import akka.actor.Status.Failure
class HttpConsumer(producer: ActorRef) extends Consumer {
def endpointUri = "jetty:http://0.0.0.0:8875/"
def receive = {
case msg producer forward msg
}
}
class HttpProducer(transformer: ActorRef) extends Actor with Producer {
def endpointUri = "jetty://http://akka.io/?bridgeEndpoint=true"
override def transformOutgoingMessage(msg: Any) = msg match {
case msg: CamelMessage msg.addHeaders(msg.headers(Set(Exchange.HTTP_PATH)))
}
override def routeResponse(msg: Any) { transformer forward msg }
}
class HttpTransformer extends Actor {
def receive = {
case msg: CamelMessage sender ! (msg.mapBody { body: Array[Byte] new String(body).replaceAll("Akka ", "AKKA ") })
case msg: Failure sender ! msg
}
}
// Create the actors. this can be done in a Boot class so you can
// run the example in the MicroKernel. just add the below three lines to your boot class.
val system = ActorSystem("some-system")
val httpTransformer = system.actorOf(Props[HttpTransformer])
val httpProducer = system.actorOf(Props(new HttpProducer(httpTransformer)))
val httpConsumer = system.actorOf(Props(new HttpConsumer(httpProducer)))
//#HttpExample
}
}

View file

@ -0,0 +1,104 @@
package docs.camel
import akka.actor.{ Props, ActorSystem }
import akka.camel.CamelExtension
import language.postfixOps
import akka.util.Timeout
object Introduction {
def foo = {
//#Consumer-mina
import akka.camel.{ CamelMessage, Consumer }
class MyEndpoint extends Consumer {
def endpointUri = "mina:tcp://localhost:6200?textline=true"
def receive = {
case msg: CamelMessage { /* ... */ }
case _ { /* ... */ }
}
}
// start and expose actor via tcp
import akka.actor.{ ActorSystem, Props }
val system = ActorSystem("some-system")
val mina = system.actorOf(Props[MyEndpoint])
//#Consumer-mina
}
def bar = {
//#Consumer
import akka.camel.{ CamelMessage, Consumer }
class MyEndpoint extends Consumer {
def endpointUri = "jetty:http://localhost:8877/example"
def receive = {
case msg: CamelMessage { /* ... */ }
case _ { /* ... */ }
}
}
//#Consumer
}
def baz = {
//#Producer
import akka.actor.Actor
import akka.camel.{ Producer, Oneway }
import akka.actor.{ ActorSystem, Props }
class Orders extends Actor with Producer with Oneway {
def endpointUri = "jms:queue:Orders"
}
val sys = ActorSystem("some-system")
val orders = sys.actorOf(Props[Orders])
orders ! <order amount="100" currency="PLN" itemId="12345"/>
//#Producer
}
{
//#CamelExtension
val system = ActorSystem("some-system")
val camel = CamelExtension(system)
val camelContext = camel.context
val producerTemplate = camel.template
//#CamelExtension
}
{
//#CamelExtensionAddComponent
// import org.apache.activemq.camel.component.ActiveMQComponent
val system = ActorSystem("some-system")
val camel = CamelExtension(system)
val camelContext = camel.context
// camelContext.addComponent("activemq", ActiveMQComponent.activeMQComponent("vm://localhost?broker.persistent=false"))
//#CamelExtensionAddComponent
}
{
//#CamelActivation
import akka.camel.{ CamelMessage, Consumer }
import scala.concurrent.util.duration._
class MyEndpoint extends Consumer {
def endpointUri = "mina:tcp://localhost:6200?textline=true"
def receive = {
case msg: CamelMessage { /* ... */ }
case _ { /* ... */ }
}
}
val system = ActorSystem("some-system")
val camel = CamelExtension(system)
val actorRef = system.actorOf(Props[MyEndpoint])
// get a future reference to the activation of the endpoint of the Consumer Actor
val activationFuture = camel.activationFutureFor(actorRef)(timeout = 10 seconds, executor = system.dispatcher)
//#CamelActivation
//#CamelDeactivation
system.stop(actorRef)
// get a future reference to the deactivation of the endpoint of the Consumer Actor
val deactivationFuture = camel.deactivationFutureFor(actorRef)(timeout = 10 seconds, executor = system.dispatcher)
//#CamelDeactivation
}
}

View file

@ -0,0 +1,128 @@
package docs.camel
import akka.camel.CamelExtension
import language.postfixOps
object Producers {
object Sample1 {
//#Producer1
import akka.actor.Actor
import akka.actor.{ Props, ActorSystem }
import akka.camel.{ Producer, CamelMessage }
import akka.util.Timeout
class Producer1 extends Actor with Producer {
def endpointUri = "http://localhost:8080/news"
}
//#Producer1
//#AskProducer
import akka.pattern.ask
import scala.concurrent.util.duration._
implicit val timeout = Timeout(10 seconds)
val system = ActorSystem("some-system")
val producer = system.actorOf(Props[Producer1])
val future = producer.ask("some request").mapTo[CamelMessage]
//#AskProducer
}
object Sample2 {
//#RouteResponse
import akka.actor.{ Actor, ActorRef }
import akka.camel.{ Producer, CamelMessage }
import akka.actor.{ Props, ActorSystem }
class ResponseReceiver extends Actor {
def receive = {
case msg: CamelMessage
// do something with the forwarded response
}
}
class Forwarder(uri: String, target: ActorRef) extends Actor with Producer {
def endpointUri = uri
override def routeResponse(msg: Any) { target forward msg }
}
val system = ActorSystem("some-system")
val receiver = system.actorOf(Props[ResponseReceiver])
val forwardResponse = system.actorOf(Props(new Forwarder("http://localhost:8080/news/akka", receiver)))
// the Forwarder sends out a request to the web page and forwards the response to
// the ResponseReceiver
forwardResponse ! "some request"
//#RouteResponse
}
object Sample3 {
//#TransformOutgoingMessage
import akka.actor.Actor
import akka.camel.{ Producer, CamelMessage }
class Transformer(uri: String) extends Actor with Producer {
def endpointUri = uri
def upperCase(msg: CamelMessage) = msg.mapBody {
body: String body.toUpperCase
}
override def transformOutgoingMessage(msg: Any) = msg match {
case msg: CamelMessage upperCase(msg)
}
}
//#TransformOutgoingMessage
}
object Sample4 {
//#Oneway
import akka.actor.{ Actor, Props, ActorSystem }
import akka.camel.Producer
class OnewaySender(uri: String) extends Actor with Producer {
def endpointUri = uri
override def oneway: Boolean = true
}
val system = ActorSystem("some-system")
val producer = system.actorOf(Props(new OnewaySender("activemq:FOO.BAR")))
producer ! "Some message"
//#Oneway
}
object Sample5 {
//#Correlate
import akka.camel.{ Producer, CamelMessage }
import akka.actor.Actor
import akka.actor.{ Props, ActorSystem }
class Producer2 extends Actor with Producer {
def endpointUri = "activemq:FOO.BAR"
}
val system = ActorSystem("some-system")
val producer = system.actorOf(Props[Producer2])
producer ! CamelMessage("bar", Map(CamelMessage.MessageExchangeId -> "123"))
//#Correlate
}
object Sample6 {
//#ProducerTemplate
import akka.actor.Actor
class MyActor extends Actor {
def receive = {
case msg
val template = CamelExtension(context.system).template
template.sendBody("direct:news", msg)
}
}
//#ProducerTemplate
}
object Sample7 {
//#RequestProducerTemplate
import akka.actor.Actor
class MyActor extends Actor {
def receive = {
case msg
val template = CamelExtension(context.system).template
sender ! template.requestBody("direct:news", msg)
}
}
//#RequestProducerTemplate
}
}

View file

@ -0,0 +1,47 @@
package docs.camel
object PublishSubscribe {
{
//#PubSub
import akka.actor.{ Actor, ActorRef, ActorSystem, Props }
import akka.camel.{ Producer, CamelMessage, Consumer }
class Subscriber(name: String, uri: String) extends Actor with Consumer {
def endpointUri = uri
def receive = {
case msg: CamelMessage println("%s received: %s" format (name, msg.body))
}
}
class Publisher(name: String, uri: String) extends Actor with Producer {
def endpointUri = uri
// one-way communication with JMS
override def oneway = true
}
class PublisherBridge(uri: String, publisher: ActorRef) extends Actor with Consumer {
def endpointUri = uri
def receive = {
case msg: CamelMessage {
publisher ! msg.bodyAs[String]
sender ! ("message published")
}
}
}
// Add below to a Boot class
// Setup publish/subscribe example
val system = ActorSystem("some-system")
val jmsUri = "jms:topic:test"
val jmsSubscriber1 = system.actorOf(Props(new Subscriber("jms-subscriber-1", jmsUri)))
val jmsSubscriber2 = system.actorOf(Props(new Subscriber("jms-subscriber-2", jmsUri)))
val jmsPublisher = system.actorOf(Props(new Publisher("jms-publisher", jmsUri)))
val jmsPublisherBridge = system.actorOf(Props(new PublisherBridge("jetty:http://0.0.0.0:8877/camel/pub/jms", jmsPublisher)))
//#PubSub
}
}

View file

@ -0,0 +1,30 @@
package docs.camel
object QuartzExample {
//#Quartz
import akka.actor.{ ActorSystem, Props }
import akka.camel.{ Consumer }
class MyQuartzActor extends Consumer {
def endpointUri = "quartz://example?cron=0/2+*+*+*+*+?"
def receive = {
case msg println("==============> received %s " format msg)
} // end receive
} // end MyQuartzActor
object MyQuartzActor {
def main(str: Array[String]) {
val system = ActorSystem("my-quartz-system")
system.actorOf(Props[MyQuartzActor])
} // end main
} // end MyQuartzActor
//#Quartz
}

View file

@ -0,0 +1,73 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.dataflow
import language.postfixOps
import scala.concurrent.util.duration._
import scala.concurrent.{ Await, Future, Promise }
import org.scalatest.WordSpec
import org.scalatest.matchers.MustMatchers
import scala.util.{ Try, Failure, Success }
class DataflowDocSpec extends WordSpec with MustMatchers {
//#import-akka-dataflow
import akka.dataflow._ //to get the flow method and implicit conversions
//#import-akka-dataflow
//#import-global-implicit
import scala.concurrent.ExecutionContext.Implicits.global
//#import-global-implicit
"demonstrate flow using hello world" in {
def println[T](any: Try[T]): Unit = any.get must be === "Hello world!"
//#simplest-hello-world
flow { "Hello world!" } onComplete println
//#simplest-hello-world
//#nested-hello-world-a
flow {
val f1 = flow { "Hello" }
f1() + " world!"
} onComplete println
//#nested-hello-world-a
//#nested-hello-world-b
flow {
val f1 = flow { "Hello" }
val f2 = flow { "world!" }
f1() + " " + f2()
} onComplete println
//#nested-hello-world-b
}
"demonstrate the use of dataflow variables" in {
def println[T](any: Try[T]): Unit = any.get must be === 20
//#dataflow-variable-a
flow {
val v1, v2 = Promise[Int]()
// v1 will become the value of v2 + 10 when v2 gets a value
v1 << v2() + 10
v2 << flow { 5 } // As you can see, no blocking!
v1() + v2()
} onComplete println
//#dataflow-variable-a
}
"demonstrate the difference between for and flow" in {
def println[T](any: Try[T]): Unit = any.get must be === 2
//#for-vs-flow
val f1, f2 = Future { 1 }
val usingFor = for { v1 f1; v2 f2 } yield v1 + v2
val usingFlow = flow { f1() + f2() }
usingFor onComplete println
usingFlow onComplete println
//#for-vs-flow
}
}

View file

@ -0,0 +1,228 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.dispatcher
import language.postfixOps
import org.scalatest.{ BeforeAndAfterAll, WordSpec }
import org.scalatest.matchers.MustMatchers
import akka.testkit.AkkaSpec
import akka.event.Logging
import akka.event.LoggingAdapter
import scala.concurrent.util.duration._
import akka.actor.{ Props, Actor, PoisonPill, ActorSystem }
object DispatcherDocSpec {
val config = """
//#my-dispatcher-config
my-dispatcher {
# Dispatcher is the name of the event-based dispatcher
type = Dispatcher
# What kind of ExecutionService to use
executor = "fork-join-executor"
# Configuration for the fork join pool
fork-join-executor {
# Min number of threads to cap factor-based parallelism number to
parallelism-min = 2
# Parallelism (threads) ... ceil(available processors * factor)
parallelism-factor = 2.0
# Max number of threads to cap factor-based parallelism number to
parallelism-max = 10
}
# Throughput defines the maximum number of messages to be
# processed per actor before the thread jumps to the next actor.
# Set to 1 for as fair as possible.
throughput = 100
}
//#my-dispatcher-config
//#my-thread-pool-dispatcher-config
my-thread-pool-dispatcher {
# Dispatcher is the name of the event-based dispatcher
type = Dispatcher
# What kind of ExecutionService to use
executor = "thread-pool-executor"
# Configuration for the thread pool
thread-pool-executor {
# minimum number of threads to cap factor-based core number to
core-pool-size-min = 2
# No of core threads ... ceil(available processors * factor)
core-pool-size-factor = 2.0
# maximum number of threads to cap factor-based number to
core-pool-size-max = 10
}
# Throughput defines the maximum number of messages to be
# processed per actor before the thread jumps to the next actor.
# Set to 1 for as fair as possible.
throughput = 100
}
//#my-thread-pool-dispatcher-config
//#my-pinned-dispatcher-config
my-pinned-dispatcher {
executor = "thread-pool-executor"
type = PinnedDispatcher
}
//#my-pinned-dispatcher-config
//#my-bounded-config
my-dispatcher-bounded-queue {
type = Dispatcher
executor = "thread-pool-executor"
thread-pool-executor {
core-pool-size-factor = 8.0
max-pool-size-factor = 16.0
}
# Specifies the bounded capacity of the mailbox queue
mailbox-capacity = 100
throughput = 3
}
//#my-bounded-config
//#my-balancing-config
my-balancing-dispatcher {
type = BalancingDispatcher
executor = "thread-pool-executor"
thread-pool-executor {
core-pool-size-factor = 8.0
max-pool-size-factor = 16.0
}
}
//#my-balancing-config
//#prio-dispatcher-config
prio-dispatcher {
mailbox-type = "docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"
}
//#prio-dispatcher-config
//#prio-dispatcher-config-java
prio-dispatcher-java {
mailbox-type = "docs.dispatcher.DispatcherDocTestBase$MyPrioMailbox"
//Other dispatcher configuration goes here
}
//#prio-dispatcher-config-java
"""
//#prio-mailbox
import akka.dispatch.PriorityGenerator
import akka.dispatch.UnboundedPriorityMailbox
import com.typesafe.config.Config
// We inherit, in this case, from UnboundedPriorityMailbox
// and seed it with the priority generator
class MyPrioMailbox(settings: ActorSystem.Settings, config: Config) extends UnboundedPriorityMailbox(
// Create a new PriorityGenerator, lower prio means more important
PriorityGenerator {
// 'highpriority messages should be treated first if possible
case 'highpriority 0
// 'lowpriority messages should be treated last if possible
case 'lowpriority 2
// PoisonPill when no other left
case PoisonPill 3
// We default to 1, which is in between high and low
case otherwise 1
})
//#prio-mailbox
class MyActor extends Actor {
def receive = {
case x
}
}
//#mailbox-implementation-example
class MyUnboundedMailbox extends akka.dispatch.MailboxType {
import akka.actor.{ ActorRef, ActorSystem }
import com.typesafe.config.Config
import java.util.concurrent.ConcurrentLinkedQueue
import akka.dispatch.{
Envelope,
MessageQueue,
QueueBasedMessageQueue,
UnboundedMessageQueueSemantics
}
// This constructor signature must exist, it will be called by Akka
def this(settings: ActorSystem.Settings, config: Config) = this()
// The create method is called to create the MessageQueue
final override def create(owner: Option[ActorRef], system: Option[ActorSystem]): MessageQueue =
new QueueBasedMessageQueue with UnboundedMessageQueueSemantics {
final val queue = new ConcurrentLinkedQueue[Envelope]()
}
}
//#mailbox-implementation-example
}
class DispatcherDocSpec extends AkkaSpec(DispatcherDocSpec.config) {
import DispatcherDocSpec.MyActor
"defining dispatcher" in {
val context = system
//#defining-dispatcher
import akka.actor.Props
val myActor =
context.actorOf(Props[MyActor].withDispatcher("my-dispatcher"), "myactor1")
//#defining-dispatcher
}
"defining dispatcher with bounded queue" in {
val dispatcher = system.dispatchers.lookup("my-dispatcher-bounded-queue")
}
"defining pinned dispatcher" in {
val context = system
//#defining-pinned-dispatcher
val myActor =
context.actorOf(Props[MyActor].withDispatcher("my-pinned-dispatcher"), "myactor2")
//#defining-pinned-dispatcher
}
"defining priority dispatcher" in {
//#prio-dispatcher
// We create a new Actor that just prints out what it processes
val a = system.actorOf(
Props(new Actor {
val log: LoggingAdapter = Logging(context.system, this)
self ! 'lowpriority
self ! 'lowpriority
self ! 'highpriority
self ! 'pigdog
self ! 'pigdog2
self ! 'pigdog3
self ! 'highpriority
self ! PoisonPill
def receive = {
case x log.info(x.toString)
}
}).withDispatcher("prio-dispatcher"))
/*
Logs:
'highpriority
'highpriority
'pigdog
'pigdog2
'pigdog3
'lowpriority
'lowpriority
*/
//#prio-dispatcher
awaitCond(a.isTerminated, 5 seconds)
}
"defining balancing dispatcher" in {
val dispatcher = system.dispatchers.lookup("my-balancing-dispatcher")
}
}

View file

@ -0,0 +1,99 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.event
import akka.testkit.AkkaSpec
import akka.actor.Actor
import akka.actor.Props
object LoggingDocSpec {
//#my-actor
import akka.event.Logging
class MyActor extends Actor {
val log = Logging(context.system, this)
override def preStart() = {
log.debug("Starting")
}
override def preRestart(reason: Throwable, message: Option[Any]) {
log.error(reason, "Restarting due to [{}] when processing [{}]",
reason.getMessage, message.getOrElse(""))
}
def receive = {
case "test" log.info("Received test")
case x log.warning("Received unknown message: {}", x)
}
}
//#my-actor
//#my-event-listener
import akka.event.Logging.InitializeLogger
import akka.event.Logging.LoggerInitialized
import akka.event.Logging.Error
import akka.event.Logging.Warning
import akka.event.Logging.Info
import akka.event.Logging.Debug
class MyEventListener extends Actor {
def receive = {
case InitializeLogger(_) sender ! LoggerInitialized
case Error(cause, logSource, logClass, message) // ...
case Warning(logSource, logClass, message) // ...
case Info(logSource, logClass, message) // ...
case Debug(logSource, logClass, message) // ...
}
}
//#my-event-listener
//#my-source
import akka.event.LogSource
import akka.actor.ActorSystem
object MyType {
implicit val logSource: LogSource[AnyRef] = new LogSource[AnyRef] {
def genString(o: AnyRef): String = o.getClass.getName
override def getClazz(o: AnyRef): Class[_] = o.getClass
}
}
class MyType(system: ActorSystem) {
import MyType._
import akka.event.Logging
val log = Logging(system, this)
}
//#my-source
}
class LoggingDocSpec extends AkkaSpec {
import LoggingDocSpec.MyActor
"use a logging actor" in {
val myActor = system.actorOf(Props(new MyActor))
myActor ! "test"
}
"allow registration to dead letters" in {
//#deadletters
import akka.actor.{ Actor, DeadLetter, Props }
val listener = system.actorOf(Props(new Actor {
def receive = {
case d: DeadLetter println(d)
}
}))
system.eventStream.subscribe(listener, classOf[DeadLetter])
//#deadletters
}
"demonstrate logging more arguments" in {
//#array
val args = Array("The", "brown", "fox", "jumps", 42)
system.log.debug("five parameters: {}, {}, {}, {}, {}", args)
//#array
}
}

View file

@ -0,0 +1,91 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.extension
import java.util.concurrent.atomic.AtomicLong
import akka.actor.Actor
import akka.testkit.AkkaSpec
//#extension
import akka.actor.Extension
class CountExtensionImpl extends Extension {
//Since this Extension is a shared instance
// per ActorSystem we need to be threadsafe
private val counter = new AtomicLong(0)
//This is the operation this Extension provides
def increment() = counter.incrementAndGet()
}
//#extension
//#extensionid
import akka.actor.ExtensionId
import akka.actor.ExtensionIdProvider
import akka.actor.ExtendedActorSystem
object CountExtension
extends ExtensionId[CountExtensionImpl]
with ExtensionIdProvider {
//The lookup method is required by ExtensionIdProvider,
// so we return ourselves here, this allows us
// to configure our extension to be loaded when
// the ActorSystem starts up
override def lookup = CountExtension
//This method will be called by Akka
// to instantiate our Extension
override def createExtension(system: ExtendedActorSystem) = new CountExtensionImpl
}
//#extensionid
object ExtensionDocSpec {
val config = """
//#config
akka {
extensions = ["docs.extension.CountExtension"]
}
//#config
"""
//#extension-usage-actor
class MyActor extends Actor {
def receive = {
case someMessage
CountExtension(context.system).increment()
}
}
//#extension-usage-actor
//#extension-usage-actor-trait
trait Counting { self: Actor
def increment() = CountExtension(context.system).increment()
}
class MyCounterActor extends Actor with Counting {
def receive = {
case someMessage increment()
}
}
//#extension-usage-actor-trait
}
class ExtensionDocSpec extends AkkaSpec(ExtensionDocSpec.config) {
import ExtensionDocSpec._
"demonstrate how to create an extension in Scala" in {
//#extension-usage
CountExtension(system).increment
//#extension-usage
}
"demonstrate how to lookup a configured extension in Scala" in {
//#extension-lookup
system.extension(CountExtension)
//#extension-lookup
}
}

View file

@ -0,0 +1,78 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.extension
//#imports
import akka.actor.Extension
import akka.actor.ExtensionId
import akka.actor.ExtensionIdProvider
import akka.actor.ExtendedActorSystem
import scala.concurrent.util.Duration
import com.typesafe.config.Config
import java.util.concurrent.TimeUnit
//#imports
import akka.actor.Actor
import akka.testkit.AkkaSpec
//#extension
class SettingsImpl(config: Config) extends Extension {
val DbUri: String = config.getString("myapp.db.uri")
val CircuitBreakerTimeout: Duration = Duration(config.getMilliseconds("myapp.circuit-breaker.timeout"), TimeUnit.MILLISECONDS)
}
//#extension
//#extensionid
object Settings extends ExtensionId[SettingsImpl] with ExtensionIdProvider {
override def lookup = Settings
override def createExtension(system: ExtendedActorSystem) = new SettingsImpl(system.settings.config)
}
//#extensionid
object SettingsExtensionDocSpec {
val config = """
//#config
myapp {
db {
uri = "mongodb://example1.com:27017,example2.com:27017"
}
circuit-breaker {
timeout = 30 seconds
}
}
//#config
"""
//#extension-usage-actor
class MyActor extends Actor {
val settings = Settings(context.system)
val connection = connect(settings.DbUri, settings.CircuitBreakerTimeout)
//#extension-usage-actor
def receive = {
case someMessage
}
def connect(dbUri: String, circuitBreakerTimeout: Duration) = {
"dummy"
}
}
}
class SettingsExtensionDocSpec extends AkkaSpec(SettingsExtensionDocSpec.config) {
"demonstrate how to create application specific settings extension in Scala" in {
//#extension-usage
val dbUri = Settings(system).DbUri
val circuitBreakerTimeout = Settings(system).CircuitBreakerTimeout
//#extension-usage
}
}

View file

@ -0,0 +1,393 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.future
import language.postfixOps
import akka.testkit._
import akka.actor.{ Actor, Props }
import akka.actor.Status
import akka.util.Timeout
import scala.concurrent.util.duration._
import java.lang.IllegalStateException
import scala.concurrent.{ Await, ExecutionContext, Future, Promise }
import scala.util.{ Failure, Success }
object FutureDocSpec {
class MyActor extends Actor {
def receive = {
case x: String sender ! x.toUpperCase
case x: Int if x < 0 sender ! Status.Failure(new ArithmeticException("Negative values not supported"))
case x: Int sender ! x
}
}
case object GetNext
class OddActor extends Actor {
var n = 1
def receive = {
case GetNext
sender ! n
n += 2
}
}
}
class FutureDocSpec extends AkkaSpec {
import FutureDocSpec._
import system.dispatcher
"demonstrate usage custom ExecutionContext" in {
val yourExecutorServiceGoesHere = java.util.concurrent.Executors.newSingleThreadExecutor()
//#diy-execution-context
import scala.concurrent.{ ExecutionContext, Promise }
implicit val ec = ExecutionContext.fromExecutorService(yourExecutorServiceGoesHere)
// Do stuff with your brand new shiny ExecutionContext
val f = Promise.successful("foo")
// Then shut your ExecutionContext down at some
// appropriate place in your program/application
ec.shutdown()
//#diy-execution-context
}
"demonstrate usage of blocking from actor" in {
val actor = system.actorOf(Props[MyActor])
val msg = "hello"
//#ask-blocking
import scala.concurrent.Await
import akka.pattern.ask
import akka.util.Timeout
import scala.concurrent.util.duration._
implicit val timeout = Timeout(5 seconds)
val future = actor ? msg // enabled by the ask import
val result = Await.result(future, timeout.duration).asInstanceOf[String]
//#ask-blocking
result must be("HELLO")
}
"demonstrate usage of mapTo" in {
val actor = system.actorOf(Props[MyActor])
val msg = "hello"
implicit val timeout = Timeout(5 seconds)
//#map-to
import scala.concurrent.Future
import akka.pattern.ask
val future: Future[String] = ask(actor, msg).mapTo[String]
//#map-to
Await.result(future, timeout.duration) must be("HELLO")
}
"demonstrate usage of simple future eval" in {
//#future-eval
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.util.duration._
val future = Future {
"Hello" + "World"
}
val result = Await.result(future, 1 second)
//#future-eval
result must be("HelloWorld")
}
"demonstrate usage of map" in {
//#map
val f1 = Future {
"Hello" + "World"
}
val f2 = f1 map { x
x.length
}
val result = Await.result(f2, 1 second)
result must be(10)
f1.value must be(Some(Success("HelloWorld")))
//#map
}
"demonstrate wrong usage of nested map" in {
//#wrong-nested-map
val f1 = Future {
"Hello" + "World"
}
val f2 = Future.successful(3)
val f3 = f1 map { x
f2 map { y
x.length * y
}
}
//#wrong-nested-map
Await.ready(f3, 1 second)
}
"demonstrate usage of flatMap" in {
//#flat-map
val f1 = Future {
"Hello" + "World"
}
val f2 = Future.successful(3)
val f3 = f1 flatMap { x
f2 map { y
x.length * y
}
}
val result = Await.result(f3, 1 second)
result must be(30)
//#flat-map
}
"demonstrate usage of filter" in {
//#filter
val future1 = Future.successful(4)
val future2 = future1.filter(_ % 2 == 0)
val result = Await.result(future2, 1 second)
result must be(4)
val failedFilter = future1.filter(_ % 2 == 1).recover {
case m: NoSuchElementException 0 //When filter fails, it will have a java.util.NoSuchElementException
}
val result2 = Await.result(failedFilter, 1 second)
result2 must be(0) //Can only be 0 when there was a MatchError
//#filter
}
"demonstrate usage of for comprehension" in {
//#for-comprehension
val f = for {
a Future(10 / 2) // 10 / 2 = 5
b Future(a + 1) // 5 + 1 = 6
c Future(a - 1) // 5 - 1 = 4
if c > 3 // Future.filter
} yield b * c // 6 * 4 = 24
// Note that the execution of futures a, b, and c
// are not done in parallel.
val result = Await.result(f, 1 second)
result must be(24)
//#for-comprehension
}
"demonstrate wrong way of composing" in {
val actor1 = system.actorOf(Props[MyActor])
val actor2 = system.actorOf(Props[MyActor])
val actor3 = system.actorOf(Props[MyActor])
val msg1 = 1
val msg2 = 2
implicit val timeout = Timeout(5 seconds)
import scala.concurrent.Await
import akka.pattern.ask
//#composing-wrong
val f1 = ask(actor1, msg1)
val f2 = ask(actor2, msg2)
val a = Await.result(f1, 1 second).asInstanceOf[Int]
val b = Await.result(f2, 1 second).asInstanceOf[Int]
val f3 = ask(actor3, (a + b))
val result = Await.result(f3, 1 second).asInstanceOf[Int]
//#composing-wrong
result must be(3)
}
"demonstrate composing" in {
val actor1 = system.actorOf(Props[MyActor])
val actor2 = system.actorOf(Props[MyActor])
val actor3 = system.actorOf(Props[MyActor])
val msg1 = 1
val msg2 = 2
implicit val timeout = Timeout(5 seconds)
import scala.concurrent.Await
import akka.pattern.ask
//#composing
val f1 = ask(actor1, msg1)
val f2 = ask(actor2, msg2)
val f3 = for {
a f1.mapTo[Int]
b f2.mapTo[Int]
c ask(actor3, (a + b)).mapTo[Int]
} yield c
val result = Await.result(f3, 1 second).asInstanceOf[Int]
//#composing
result must be(3)
}
"demonstrate usage of sequence with actors" in {
implicit val timeout = Timeout(5 seconds)
val oddActor = system.actorOf(Props[OddActor])
//#sequence-ask
// oddActor returns odd numbers sequentially from 1 as a List[Future[Int]]
val listOfFutures = List.fill(100)(akka.pattern.ask(oddActor, GetNext).mapTo[Int])
// now we have a Future[List[Int]]
val futureList = Future.sequence(listOfFutures)
// Find the sum of the odd numbers
val oddSum = Await.result(futureList.map(_.sum), 1 second).asInstanceOf[Int]
oddSum must be(10000)
//#sequence-ask
}
"demonstrate usage of sequence" in {
//#sequence
val futureList = Future.sequence((1 to 100).toList.map(x Future(x * 2 - 1)))
val oddSum = Await.result(futureList.map(_.sum), 1 second).asInstanceOf[Int]
oddSum must be(10000)
//#sequence
}
"demonstrate usage of traverse" in {
//#traverse
val futureList = Future.traverse((1 to 100).toList)(x Future(x * 2 - 1))
val oddSum = Await.result(futureList.map(_.sum), 1 second).asInstanceOf[Int]
oddSum must be(10000)
//#traverse
}
"demonstrate usage of fold" in {
//#fold
val futures = for (i 1 to 1000) yield Future(i * 2) // Create a sequence of Futures
val futureSum = Future.fold(futures)(0)(_ + _)
Await.result(futureSum, 1 second) must be(1001000)
//#fold
}
"demonstrate usage of reduce" in {
//#reduce
val futures = for (i 1 to 1000) yield Future(i * 2) // Create a sequence of Futures
val futureSum = Future.reduce(futures)(_ + _)
Await.result(futureSum, 1 second) must be(1001000)
//#reduce
}
"demonstrate usage of recover" in {
implicit val timeout = Timeout(5 seconds)
val actor = system.actorOf(Props[MyActor])
val msg1 = -1
//#recover
val future = akka.pattern.ask(actor, msg1) recover {
case e: ArithmeticException 0
}
//#recover
Await.result(future, 1 second) must be(0)
}
"demonstrate usage of recoverWith" in {
implicit val timeout = Timeout(5 seconds)
val actor = system.actorOf(Props[MyActor])
val msg1 = -1
//#try-recover
val future = akka.pattern.ask(actor, msg1) recoverWith {
case e: ArithmeticException Future.successful(0)
case foo: IllegalArgumentException Future.failed[Int](new IllegalStateException("All br0ken!"))
}
//#try-recover
Await.result(future, 1 second) must be(0)
}
"demonstrate usage of zip" in {
val future1 = Future { "foo" }
val future2 = Future { "bar" }
//#zip
val future3 = future1 zip future2 map { case (a, b) a + " " + b }
//#zip
Await.result(future3, 1 second) must be("foo bar")
}
"demonstrate usage of andThen" in {
def loadPage(s: String) = s
val url = "foo bar"
def log(cause: Throwable) = ()
def watchSomeTV = ()
//#and-then
val result = Future { loadPage(url) } andThen {
case Failure(exception) log(exception)
} andThen {
case _ watchSomeTV
}
//#and-then
Await.result(result, 1 second) must be("foo bar")
}
"demonstrate usage of fallbackTo" in {
val future1 = Future { "foo" }
val future2 = Future { "bar" }
val future3 = Future { "pigdog" }
//#fallback-to
val future4 = future1 fallbackTo future2 fallbackTo future3
//#fallback-to
Await.result(future4, 1 second) must be("foo")
}
"demonstrate usage of onSuccess & onFailure & onComplete" in {
{
val future = Future { "foo" }
//#onSuccess
future onSuccess {
case "bar" println("Got my bar alright!")
case x: String println("Got some random string: " + x)
}
//#onSuccess
Await.result(future, 1 second) must be("foo")
}
{
val future = Future.failed[String](new IllegalStateException("OHNOES"))
//#onFailure
future onFailure {
case ise: IllegalStateException if ise.getMessage == "OHNOES"
//OHNOES! We are in deep trouble, do something!
case e: Exception
//Do something else
}
//#onFailure
}
{
val future = Future { "foo" }
def doSomethingOnSuccess(r: String) = ()
def doSomethingOnFailure(t: Throwable) = ()
//#onComplete
future onComplete {
case Success(result) doSomethingOnSuccess(result)
case Failure(failure) doSomethingOnFailure(failure)
}
//#onComplete
Await.result(future, 1 second) must be("foo")
}
}
"demonstrate usage of Future.successful & Future.failed" in {
//#successful
val future = Future.successful("Yay!")
//#successful
//#failed
val otherFuture = Future.failed[String](new IllegalArgumentException("Bang!"))
//#failed
Await.result(future, 1 second) must be("Yay!")
intercept[IllegalArgumentException] { Await.result(otherFuture, 1 second) }
}
"demonstrate usage of pattern.after" in {
//#after
import akka.pattern.after
val delayed = after(200 millis, using = system.scheduler)(Future.failed(
new IllegalStateException("OHNOES")))
val future = Future { Thread.sleep(1000); "foo" }
val result = future either delayed
//#after
intercept[IllegalStateException] { Await.result(result, 2 second) }
}
}

View file

@ -0,0 +1,84 @@
/**
* Copyright (C) 2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.io
//#imports
import akka.actor._
import akka.util.{ ByteString, ByteStringBuilder, ByteIterator }
//#imports
abstract class BinaryDecoding {
//#decoding
implicit val byteOrder = java.nio.ByteOrder.BIG_ENDIAN
val FrameDecoder = for {
frameLenBytes IO.take(4)
frameLen = frameLenBytes.iterator.getInt
frame IO.take(frameLen)
} yield {
val in = frame.iterator
val n = in.getInt
val m = in.getInt
val a = Array.newBuilder[Short]
val b = Array.newBuilder[Long]
for (i 1 to n) {
a += in.getShort
b += in.getInt
}
val data = Array.ofDim[Double](m)
in.getDoubles(data)
(a.result, b.result, data)
}
//#decoding
}
abstract class RestToSeq {
implicit val byteOrder = java.nio.ByteOrder.BIG_ENDIAN
val bytes: ByteString
val in = bytes.iterator
//#rest-to-seq
val n = in.getInt
val m = in.getInt
// ... in.get...
val rest: ByteString = in.toSeq
//#rest-to-seq
}
abstract class BinaryEncoding {
//#encoding
implicit val byteOrder = java.nio.ByteOrder.BIG_ENDIAN
val a: Array[Short]
val b: Array[Long]
val data: Array[Double]
val frameBuilder = ByteString.newBuilder
val n = a.length
val m = data.length
frameBuilder.putInt(n)
frameBuilder.putInt(m)
for (i 0 to n - 1) {
frameBuilder.putShort(a(i))
frameBuilder.putLong(b(i))
}
frameBuilder.putDoubles(data)
val frame = frameBuilder.result()
//#encoding
//#sending
val socket: IO.SocketHandle
socket.write(ByteString.newBuilder.putInt(frame.length).result)
socket.write(frame)
//#sending
}

View file

@ -0,0 +1,226 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.io
import language.postfixOps
//#imports
import akka.actor._
import akka.util.{ ByteString, ByteStringBuilder }
import java.net.InetSocketAddress
//#imports
//#actor
class HttpServer(port: Int) extends Actor {
val state = IO.IterateeRef.Map.async[IO.Handle]()(context.dispatcher)
override def preStart {
IOManager(context.system) listen new InetSocketAddress(port)
}
def receive = {
case IO.NewClient(server)
val socket = server.accept()
state(socket) flatMap (_ HttpServer.processRequest(socket))
case IO.Read(socket, bytes)
state(socket)(IO Chunk bytes)
case IO.Closed(socket, cause)
state(socket)(IO EOF)
state -= socket
}
}
//#actor
//#actor-companion
object HttpServer {
import HttpIteratees._
def processRequest(socket: IO.SocketHandle): IO.Iteratee[Unit] =
IO repeat {
for {
request readRequest
} yield {
val rsp = request match {
case Request("GET", "ping" :: Nil, _, _, headers, _)
OKResponse(ByteString("<p>pong</p>"),
request.headers.exists { case Header(n, v) n.toLowerCase == "connection" && v.toLowerCase == "keep-alive" })
case req
OKResponse(ByteString("<p>" + req.toString + "</p>"),
request.headers.exists { case Header(n, v) n.toLowerCase == "connection" && v.toLowerCase == "keep-alive" })
}
socket write OKResponse.bytes(rsp).compact
if (!rsp.keepAlive) socket.close()
}
}
}
//#actor-companion
//#request-class
case class Request(meth: String, path: List[String], query: Option[String], httpver: String, headers: List[Header], body: Option[ByteString])
case class Header(name: String, value: String)
//#request-class
//#constants
object HttpConstants {
val SP = ByteString(" ")
val HT = ByteString("\t")
val CRLF = ByteString("\r\n")
val COLON = ByteString(":")
val PERCENT = ByteString("%")
val PATH = ByteString("/")
val QUERY = ByteString("?")
}
//#constants
//#read-request
object HttpIteratees {
import HttpConstants._
def readRequest =
for {
requestLine readRequestLine
(meth, (path, query), httpver) = requestLine
headers readHeaders
body readBody(headers)
} yield Request(meth, path, query, httpver, headers, body)
//#read-request
//#read-request-line
def ascii(bytes: ByteString): String = bytes.decodeString("US-ASCII").trim
def readRequestLine =
for {
meth IO takeUntil SP
uri readRequestURI
_ IO takeUntil SP // ignore the rest
httpver IO takeUntil CRLF
} yield (ascii(meth), uri, ascii(httpver))
//#read-request-line
//#read-request-uri
def readRequestURI = IO peek 1 flatMap {
case PATH
for {
path readPath
query readQuery
} yield (path, query)
case _ sys.error("Not Implemented")
}
//#read-request-uri
//#read-path
def readPath = {
def step(segments: List[String]): IO.Iteratee[List[String]] = IO peek 1 flatMap {
case PATH IO drop 1 flatMap (_ readUriPart(pathchar) flatMap (segment step(segment :: segments)))
case _ segments match {
case "" :: rest IO Done rest.reverse
case _ IO Done segments.reverse
}
}
step(Nil)
}
//#read-path
//#read-query
def readQuery: IO.Iteratee[Option[String]] = IO peek 1 flatMap {
case QUERY IO drop 1 flatMap (_ readUriPart(querychar) map (Some(_)))
case _ IO Done None
}
//#read-query
//#read-uri-part
val alpha = Set.empty ++ ('a' to 'z') ++ ('A' to 'Z') map (_.toByte)
val digit = Set.empty ++ ('0' to '9') map (_.toByte)
val hexdigit = digit ++ (Set.empty ++ ('a' to 'f') ++ ('A' to 'F') map (_.toByte))
val subdelim = Set('!', '$', '&', '\'', '(', ')', '*', '+', ',', ';', '=') map (_.toByte)
val pathchar = alpha ++ digit ++ subdelim ++ (Set(':', '@') map (_.toByte))
val querychar = pathchar ++ (Set('/', '?') map (_.toByte))
def readUriPart(allowed: Set[Byte]): IO.Iteratee[String] = for {
str IO takeWhile allowed map ascii
pchar IO peek 1 map (_ == PERCENT)
all if (pchar) readPChar flatMap (ch readUriPart(allowed) map (str + ch + _)) else IO Done str
} yield all
def readPChar = IO take 3 map {
case Seq('%', rest @ _*) if rest forall hexdigit
java.lang.Integer.parseInt(rest map (_.toChar) mkString, 16).toChar
}
//#read-uri-part
//#read-headers
def readHeaders = {
def step(found: List[Header]): IO.Iteratee[List[Header]] = {
IO peek 2 flatMap {
case CRLF IO takeUntil CRLF flatMap (_ IO Done found)
case _ readHeader flatMap (header step(header :: found))
}
}
step(Nil)
}
def readHeader =
for {
name IO takeUntil COLON
value IO takeUntil CRLF flatMap readMultiLineValue
} yield Header(ascii(name), ascii(value))
def readMultiLineValue(initial: ByteString): IO.Iteratee[ByteString] = IO peek 1 flatMap {
case SP IO takeUntil CRLF flatMap (bytes readMultiLineValue(initial ++ bytes))
case _ IO Done initial
}
//#read-headers
//#read-body
def readBody(headers: List[Header]) =
if (headers.exists(header header.name == "Content-Length" || header.name == "Transfer-Encoding"))
IO.takeAll map (Some(_))
else
IO Done None
//#read-body
}
//#ok-response
object OKResponse {
import HttpConstants.CRLF
val okStatus = ByteString("HTTP/1.1 200 OK")
val contentType = ByteString("Content-Type: text/html; charset=utf-8")
val cacheControl = ByteString("Cache-Control: no-cache")
val date = ByteString("Date: ")
val server = ByteString("Server: Akka")
val contentLength = ByteString("Content-Length: ")
val connection = ByteString("Connection: ")
val keepAlive = ByteString("Keep-Alive")
val close = ByteString("Close")
def bytes(rsp: OKResponse) = {
new ByteStringBuilder ++=
okStatus ++= CRLF ++=
contentType ++= CRLF ++=
cacheControl ++= CRLF ++=
date ++= ByteString(new java.util.Date().toString) ++= CRLF ++=
server ++= CRLF ++=
contentLength ++= ByteString(rsp.body.length.toString) ++= CRLF ++=
connection ++= (if (rsp.keepAlive) keepAlive else close) ++= CRLF ++= CRLF ++= rsp.body result
}
}
case class OKResponse(body: ByteString, keepAlive: Boolean)
//#ok-response
//#main
object Main extends App {
val port = Option(System.getenv("PORT")) map (_.toInt) getOrElse 8080
val system = ActorSystem()
val server = system.actorOf(Props(new HttpServer(port)))
}
//#main

View file

@ -0,0 +1,16 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.pattern
// this part will not appear in the docs
//#all-of-it
class ScalaTemplate {
println("Hello, Template!")
//#uninteresting-stuff
// dont show this plumbimg
//#uninteresting-stuff
}
//#all-of-it

View file

@ -0,0 +1,60 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.remoting
import akka.actor.{ ExtendedActorSystem, ActorSystem, Actor, ActorRef }
import akka.testkit.{ AkkaSpec, ImplicitSender }
//#import
import akka.actor.{ Props, Deploy, Address, AddressFromURIString }
import akka.remote.RemoteScope
//#import
object RemoteDeploymentDocSpec {
//#sample-actor
class Echo extends Actor {
def receive = { case _ sender ! self }
}
//#sample-actor
}
class RemoteDeploymentDocSpec extends AkkaSpec("""
akka.actor.provider = "akka.remote.RemoteActorRefProvider"
akka.remote.netty.port = 0
""") with ImplicitSender {
import RemoteDeploymentDocSpec._
val other = ActorSystem("remote", system.settings.config)
val address = other.asInstanceOf[ExtendedActorSystem].provider.getExternalAddressFor(Address("akka", "s", "host", 1)).get
override def atTermination() { other.shutdown() }
"demonstrate programmatic deployment" in {
//#deploy
val ref = system.actorOf(Props[Echo].withDeploy(Deploy(scope = RemoteScope(address))))
//#deploy
ref.path.address must be(address)
ref ! "test"
expectMsgType[ActorRef].path.address must be(address)
}
"demonstrate address extractor" in {
//#make-address
val one = AddressFromURIString("akka://sys@host:1234")
val two = Address("akka", "sys", "host", 1234) // this gives the same
//#make-address
one must be === two
}
"demonstrate sampleActor" in {
//#sample-actor
val actor = system.actorOf(Props[Echo], "sampleActor")
actor ! "Pretty slick"
//#sample-actor
}
}

View file

@ -0,0 +1,73 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.routing
import akka.testkit.AkkaSpec
import akka.testkit.ImplicitSender
object ConsistentHashingRouterDocSpec {
//#cache-actor
import akka.actor.Actor
import akka.routing.ConsistentHashingRouter.ConsistentHashable
class Cache extends Actor {
var cache = Map.empty[String, String]
def receive = {
case Entry(key, value) cache += (key -> value)
case Get(key) sender ! cache.get(key)
case Evict(key) cache -= key
}
}
case class Evict(key: String)
case class Get(key: String) extends ConsistentHashable {
override def consistentHashKey: Any = key
}
case class Entry(key: String, value: String)
//#cache-actor
}
class ConsistentHashingRouterDocSpec extends AkkaSpec with ImplicitSender {
import ConsistentHashingRouterDocSpec._
"demonstrate usage of ConsistentHashableRouter" in {
//#consistent-hashing-router
import akka.actor.Props
import akka.routing.ConsistentHashingRouter
import akka.routing.ConsistentHashingRouter.ConsistentHashMapping
import akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope
def hashMapping: ConsistentHashMapping = {
case Evict(key) key
}
val cache = system.actorOf(Props[Cache].withRouter(ConsistentHashingRouter(10,
hashMapping = hashMapping)), name = "cache")
cache ! ConsistentHashableEnvelope(
message = Entry("hello", "HELLO"), hashKey = "hello")
cache ! ConsistentHashableEnvelope(
message = Entry("hi", "HI"), hashKey = "hi")
cache ! Get("hello")
expectMsg(Some("HELLO"))
cache ! Get("hi")
expectMsg(Some("HI"))
cache ! Evict("hi")
cache ! Get("hi")
expectMsg(None)
//#consistent-hashing-router
}
}

View file

@ -0,0 +1,29 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.routing
import RouterDocSpec.MyActor
import akka.testkit.AkkaSpec
import akka.routing.RoundRobinRouter
import akka.actor.{ ActorRef, Props, Actor }
object RouterDocSpec {
class MyActor extends Actor {
def receive = {
case _
}
}
}
class RouterDocSpec extends AkkaSpec {
import RouterDocSpec._
//#dispatchers
val router: ActorRef = system.actorOf(Props[MyActor]
.withRouter(RoundRobinRouter(5, routerDispatcher = "router")) // head will run on "router" dispatcher
.withDispatcher("workers")) // MyActor workers will run on "workers" dispatcher
//#dispatchers
}

View file

@ -0,0 +1,94 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.routing
import language.postfixOps
import akka.routing.{ ScatterGatherFirstCompletedRouter, BroadcastRouter, RandomRouter, RoundRobinRouter }
import annotation.tailrec
import akka.actor.{ Props, Actor }
import scala.concurrent.util.duration._
import akka.util.Timeout
import scala.concurrent.Await
import akka.pattern.ask
import akka.routing.SmallestMailboxRouter
case class FibonacciNumber(nbr: Int)
//#printlnActor
class PrintlnActor extends Actor {
def receive = {
case msg
println("Received message '%s' in actor %s".format(msg, self.path.name))
}
}
//#printlnActor
//#fibonacciActor
class FibonacciActor extends Actor {
def receive = {
case FibonacciNumber(nbr) sender ! fibonacci(nbr)
}
private def fibonacci(n: Int): Int = {
@tailrec
def fib(n: Int, b: Int, a: Int): Int = n match {
case 0 a
case _ fib(n - 1, a + b, b)
}
fib(n, 1, 0)
}
}
//#fibonacciActor
//#parentActor
class ParentActor extends Actor {
def receive = {
case "rrr"
//#roundRobinRouter
val roundRobinRouter =
context.actorOf(Props[PrintlnActor].withRouter(RoundRobinRouter(5)), "router")
1 to 10 foreach {
i roundRobinRouter ! i
}
//#roundRobinRouter
case "rr"
//#randomRouter
val randomRouter =
context.actorOf(Props[PrintlnActor].withRouter(RandomRouter(5)), "router")
1 to 10 foreach {
i randomRouter ! i
}
//#randomRouter
case "smr"
//#smallestMailboxRouter
val smallestMailboxRouter =
context.actorOf(Props[PrintlnActor].withRouter(SmallestMailboxRouter(5)), "router")
1 to 10 foreach {
i smallestMailboxRouter ! i
}
//#smallestMailboxRouter
case "br"
//#broadcastRouter
val broadcastRouter =
context.actorOf(Props[PrintlnActor].withRouter(BroadcastRouter(5)), "router")
broadcastRouter ! "this is a broadcast message"
//#broadcastRouter
case "sgfcr"
//#scatterGatherFirstCompletedRouter
val scatterGatherFirstCompletedRouter = context.actorOf(
Props[FibonacciActor].withRouter(ScatterGatherFirstCompletedRouter(
nrOfInstances = 5, within = 2 seconds)), "router")
implicit val timeout = Timeout(5 seconds)
val futureResult = scatterGatherFirstCompletedRouter ? FibonacciNumber(10)
val result = Await.result(futureResult, timeout.duration)
//#scatterGatherFirstCompletedRouter
println("The result of calculating Fibonacci for 10 is %d".format(result))
}
}
//#parentActor

View file

@ -0,0 +1,158 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.routing
import akka.actor.{ Actor, Props, ActorSystem, ActorLogging }
import com.typesafe.config.ConfigFactory
import akka.routing.FromConfig
import akka.routing.ConsistentHashingRouter.ConsistentHashable
import akka.testkit.AkkaSpec
import akka.testkit.ImplicitSender
object RouterWithConfigDocSpec {
val config = ConfigFactory.parseString("""
//#config-round-robin
akka.actor.deployment {
/myrouter1 {
router = round-robin
nr-of-instances = 5
}
}
//#config-round-robin
//#config-resize
akka.actor.deployment {
/myrouter2 {
router = round-robin
resizer {
lower-bound = 2
upper-bound = 15
}
}
}
//#config-resize
//#config-random
akka.actor.deployment {
/myrouter3 {
router = random
nr-of-instances = 5
}
}
//#config-random
//#config-smallest-mailbox
akka.actor.deployment {
/myrouter4 {
router = smallest-mailbox
nr-of-instances = 5
}
}
//#config-smallest-mailbox
//#config-broadcast
akka.actor.deployment {
/myrouter5 {
router = broadcast
nr-of-instances = 5
}
}
//#config-broadcast
//#config-scatter-gather
akka.actor.deployment {
/myrouter6 {
router = scatter-gather
nr-of-instances = 5
within = 10 seconds
}
}
//#config-scatter-gather
//#config-consistent-hashing
akka.actor.deployment {
/myrouter7 {
router = consistent-hashing
nr-of-instances = 5
virtual-nodes-factor = 10
}
}
//#config-consistent-hashing
""")
case class Message(nbr: Int) extends ConsistentHashable {
override def consistentHashKey = nbr
}
class ExampleActor extends Actor with ActorLogging {
def receive = {
case Message(nbr)
log.debug("Received %s in router %s".format(nbr, self.path.name))
sender ! nbr
}
}
}
class RouterWithConfigDocSpec extends AkkaSpec(RouterWithConfigDocSpec.config) with ImplicitSender {
import RouterWithConfigDocSpec._
"demonstrate configured round-robin router" in {
//#configurableRouting
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter1")
//#configurableRouting
1 to 10 foreach { i router ! Message(i) }
receiveN(10)
}
"demonstrate configured random router" in {
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter3")
1 to 10 foreach { i router ! Message(i) }
receiveN(10)
}
"demonstrate configured smallest-mailbox router" in {
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter4")
1 to 10 foreach { i router ! Message(i) }
receiveN(10)
}
"demonstrate configured broadcast router" in {
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter5")
1 to 10 foreach { i router ! Message(i) }
receiveN(5 * 10)
}
"demonstrate configured scatter-gather router" in {
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter6")
1 to 10 foreach { i router ! Message(i) }
receiveN(10)
}
"demonstrate configured consistent-hashing router" in {
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter7")
1 to 10 foreach { i router ! Message(i) }
receiveN(10)
}
"demonstrate configured round-robin router with resizer" in {
//#configurableRoutingWithResizer
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"myrouter2")
//#configurableRoutingWithResizer
1 to 10 foreach { i router ! Message(i) }
receiveN(10)
}
}

View file

@ -0,0 +1,52 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.routing
import akka.actor.{ Actor, Props, ActorSystem }
import com.typesafe.config.ConfigFactory
import akka.routing.FromConfig
case class Message(nbr: Int)
class ExampleActor extends Actor {
def receive = {
case Message(nbr) println("Received %s in router %s".format(nbr, self.path.name))
}
}
object RouterWithConfigExample extends App {
val config = ConfigFactory.parseString("""
//#config
akka.actor.deployment {
/router {
router = round-robin
nr-of-instances = 5
}
}
//#config
//#config-resize
akka.actor.deployment {
/router2 {
router = round-robin
resizer {
lower-bound = 2
upper-bound = 15
}
}
}
//#config-resize
""")
val system = ActorSystem("Example", config)
//#configurableRouting
val router = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"router")
//#configurableRouting
1 to 10 foreach { i router ! Message(i) }
//#configurableRoutingWithResizer
val router2 = system.actorOf(Props[ExampleActor].withRouter(FromConfig()),
"router2")
//#configurableRoutingWithResizer
1 to 10 foreach { i router2 ! Message(i) }
}

View file

@ -0,0 +1,53 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.routing
import akka.routing.RoundRobinRouter
import akka.actor.{ ActorRef, Props, Actor, ActorSystem }
import akka.routing.DefaultResizer
import akka.remote.routing.RemoteRouterConfig
case class Message1(nbr: Int)
class ExampleActor1 extends Actor {
def receive = {
case Message1(nbr) println("Received %s in router %s".format(nbr, self.path.name))
}
}
object RoutingProgrammaticallyExample extends App {
val system = ActorSystem("RPE")
//#programmaticRoutingNrOfInstances
val router1 = system.actorOf(Props[ExampleActor1].withRouter(
RoundRobinRouter(nrOfInstances = 5)))
//#programmaticRoutingNrOfInstances
1 to 6 foreach { i router1 ! Message1(i) }
//#programmaticRoutingRoutees
val actor1 = system.actorOf(Props[ExampleActor1])
val actor2 = system.actorOf(Props[ExampleActor1])
val actor3 = system.actorOf(Props[ExampleActor1])
val routees = Vector[ActorRef](actor1, actor2, actor3)
val router2 = system.actorOf(Props().withRouter(
RoundRobinRouter(routees = routees)))
//#programmaticRoutingRoutees
1 to 6 foreach { i router2 ! Message1(i) }
//#programmaticRoutingWithResizer
val resizer = DefaultResizer(lowerBound = 2, upperBound = 15)
val router3 = system.actorOf(Props[ExampleActor1].withRouter(
RoundRobinRouter(resizer = Some(resizer))))
//#programmaticRoutingWithResizer
1 to 6 foreach { i router3 ! Message1(i) }
//#remoteRoutees
import akka.actor.{ Address, AddressFromURIString }
val addresses = Seq(
Address("akka", "remotesys", "otherhost", 1234),
AddressFromURIString("akka://othersys@anotherhost:1234"))
val routerRemote = system.actorOf(Props[ExampleActor1].withRouter(
RemoteRouterConfig(RoundRobinRouter(5), addresses)))
//#remoteRoutees
}

View file

@ -0,0 +1,227 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
//#extract-transport
package object akka {
// needs to be inside the akka package because accessing unsupported API !
def transportOf(system: actor.ExtendedActorSystem): remote.RemoteTransport =
system.provider match {
case r: remote.RemoteActorRefProvider r.transport
case _
throw new UnsupportedOperationException(
"this method requires the RemoteActorRefProvider to be configured")
}
}
//#extract-transport
package docs.serialization {
import org.scalatest.matchers.MustMatchers
import akka.testkit._
//#imports
import akka.actor.{ ActorRef, ActorSystem }
import akka.serialization._
import com.typesafe.config.ConfigFactory
//#imports
import akka.actor.ExtensionKey
import akka.actor.ExtendedActorSystem
import akka.actor.Extension
import akka.actor.Address
import akka.remote.RemoteActorRefProvider
//#my-own-serializer
class MyOwnSerializer extends Serializer {
// This is whether "fromBinary" requires a "clazz" or not
def includeManifest: Boolean = false
// Pick a unique identifier for your Serializer,
// you've got a couple of billions to choose from,
// 0 - 16 is reserved by Akka itself
def identifier = 1234567
// "toBinary" serializes the given object to an Array of Bytes
def toBinary(obj: AnyRef): Array[Byte] = {
// Put the code that serializes the object here
//#...
Array[Byte]()
//#...
}
// "fromBinary" deserializes the given array,
// using the type hint (if any, see "includeManifest" above)
// into the optionally provided classLoader.
def fromBinary(bytes: Array[Byte],
clazz: Option[Class[_]]): AnyRef = {
// Put your code that deserializes here
//#...
null
//#...
}
}
//#my-own-serializer
trait MyOwnSerializable
case class Customer(name: String) extends MyOwnSerializable
class SerializationDocSpec extends AkkaSpec {
"demonstrate configuration of serialize messages" in {
//#serialize-messages-config
val config = ConfigFactory.parseString("""
akka {
actor {
serialize-messages = on
}
}
""")
//#serialize-messages-config
val a = ActorSystem("system", config)
a.settings.SerializeAllMessages must be(true)
a.shutdown()
}
"demonstrate configuration of serialize creators" in {
//#serialize-creators-config
val config = ConfigFactory.parseString("""
akka {
actor {
serialize-creators = on
}
}
""")
//#serialize-creators-config
val a = ActorSystem("system", config)
a.settings.SerializeAllCreators must be(true)
a.shutdown()
}
"demonstrate configuration of serializers" in {
//#serialize-serializers-config
val config = ConfigFactory.parseString("""
akka {
actor {
serializers {
java = "akka.serialization.JavaSerializer"
proto = "akka.remote.serialization.ProtobufSerializer"
myown = "docs.serialization.MyOwnSerializer"
}
}
}
""")
//#serialize-serializers-config
val a = ActorSystem("system", config)
a.shutdown()
}
"demonstrate configuration of serialization-bindings" in {
//#serialization-bindings-config
val config = ConfigFactory.parseString("""
akka {
actor {
serializers {
java = "akka.serialization.JavaSerializer"
proto = "akka.remote.serialization.ProtobufSerializer"
myown = "docs.serialization.MyOwnSerializer"
}
serialization-bindings {
"java.lang.String" = java
"docs.serialization.Customer" = java
"com.google.protobuf.Message" = proto
"docs.serialization.MyOwnSerializable" = myown
"java.lang.Boolean" = myown
}
}
}
""")
//#serialization-bindings-config
val a = ActorSystem("system", config)
SerializationExtension(a).serializerFor(classOf[String]).getClass must equal(classOf[JavaSerializer])
SerializationExtension(a).serializerFor(classOf[Customer]).getClass must equal(classOf[JavaSerializer])
SerializationExtension(a).serializerFor(classOf[java.lang.Boolean]).getClass must equal(classOf[MyOwnSerializer])
a.shutdown()
}
"demonstrate the programmatic API" in {
//#programmatic
val system = ActorSystem("example")
// Get the Serialization Extension
val serialization = SerializationExtension(system)
// Have something to serialize
val original = "woohoo"
// Find the Serializer for it
val serializer = serialization.findSerializerFor(original)
// Turn it into bytes
val bytes = serializer.toBinary(original)
// Turn it back into an object
val back = serializer.fromBinary(bytes, manifest = None)
// Voilá!
back must equal(original)
//#programmatic
system.shutdown()
}
"demonstrate serialization of ActorRefs" in {
val theActorRef: ActorRef = system.deadLetters
val theActorSystem: ActorSystem = system
//#actorref-serializer
// Serialize
// (beneath toBinary)
// If there is no transportAddress,
// it means that either this Serializer isn't called
// within a piece of code that sets it,
// so either you need to supply your own,
// or simply use the local path.
val identifier: String = Serialization.currentTransportAddress.value match {
case null theActorRef.path.toString
case address theActorRef.path.toStringWithAddress(address)
}
// Then just serialize the identifier however you like
// Deserialize
// (beneath fromBinary)
val deserializedActorRef = theActorSystem actorFor identifier
// Then just use the ActorRef
//#actorref-serializer
//#external-address
object ExternalAddress extends ExtensionKey[ExternalAddressExt]
class ExternalAddressExt(system: ExtendedActorSystem) extends Extension {
def addressFor(remoteAddr: Address): Address =
system.provider.getExternalAddressFor(remoteAddr) getOrElse
(throw new UnsupportedOperationException("cannot send to " + remoteAddr))
}
def serializeTo(ref: ActorRef, remote: Address): String =
ref.path.toStringWithAddress(ExternalAddress(theActorSystem).addressFor(remote))
//#external-address
}
"demonstrate how to do default Akka serialization of ActorRef" in {
val theActorSystem: ActorSystem = system
//#external-address-default
object ExternalAddress extends ExtensionKey[ExternalAddressExt]
class ExternalAddressExt(system: ExtendedActorSystem) extends Extension {
def addressForAkka: Address = akka.transportOf(system).address
}
def serializeAkkaDefault(ref: ActorRef): String =
ref.path.toStringWithAddress(ExternalAddress(theActorSystem).addressForAkka)
//#external-address-default
}
}
}

View file

@ -0,0 +1,47 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.testkit
//#plain-spec
import akka.actor.ActorSystem
import akka.actor.Actor
import akka.actor.Props
import akka.testkit.TestKit
import org.scalatest.WordSpec
import org.scalatest.matchers.MustMatchers
import org.scalatest.BeforeAndAfterAll
import akka.testkit.ImplicitSender
object MySpec {
class EchoActor extends Actor {
def receive = {
case x sender ! x
}
}
}
//#implicit-sender
class MySpec(_system: ActorSystem) extends TestKit(_system) with ImplicitSender
with WordSpec with MustMatchers with BeforeAndAfterAll {
//#implicit-sender
def this() = this(ActorSystem("MySpec"))
import MySpec._
override def afterAll {
system.shutdown()
}
"An Echo actor" must {
"send back messages unchanged" in {
val echo = system.actorOf(Props[EchoActor])
echo ! "hello world"
expectMsg("hello world")
}
}
}
//#plain-spec

View file

@ -0,0 +1,158 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.testkit
import language.postfixOps
//#testkit-usage
import scala.util.Random
import org.scalatest.BeforeAndAfterAll
import org.scalatest.WordSpec
import org.scalatest.matchers.ShouldMatchers
import com.typesafe.config.ConfigFactory
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Props
import akka.testkit.DefaultTimeout
import akka.testkit.ImplicitSender
import akka.testkit.TestKit
import scala.concurrent.util.duration._
/**
* a Test to show some TestKit examples
*/
class TestKitUsageSpec
extends TestKit(ActorSystem("TestKitUsageSpec",
ConfigFactory.parseString(TestKitUsageSpec.config)))
with DefaultTimeout with ImplicitSender
with WordSpec with ShouldMatchers with BeforeAndAfterAll {
import TestKitUsageSpec._
val echoRef = system.actorOf(Props(new EchoActor))
val forwardRef = system.actorOf(Props(new ForwardingActor(testActor)))
val filterRef = system.actorOf(Props(new FilteringActor(testActor)))
val randomHead = Random.nextInt(6)
val randomTail = Random.nextInt(10)
val headList = Seq().padTo(randomHead, "0")
val tailList = Seq().padTo(randomTail, "1")
val seqRef = system.actorOf(Props(new SequencingActor(testActor, headList, tailList)))
override def afterAll {
system.shutdown()
}
"An EchoActor" should {
"Respond with the same message it receives" in {
within(500 millis) {
echoRef ! "test"
expectMsg("test")
}
}
}
"A ForwardingActor" should {
"Forward a message it receives" in {
within(500 millis) {
forwardRef ! "test"
expectMsg("test")
}
}
}
"A FilteringActor" should {
"Filter all messages, except expected messagetypes it receives" in {
var messages = Seq[String]()
within(500 millis) {
filterRef ! "test"
expectMsg("test")
filterRef ! 1
expectNoMsg
filterRef ! "some"
filterRef ! "more"
filterRef ! 1
filterRef ! "text"
filterRef ! 1
receiveWhile(500 millis) {
case msg: String messages = msg +: messages
}
}
messages.length should be(3)
messages.reverse should be(Seq("some", "more", "text"))
}
}
"A SequencingActor" should {
"receive an interesting message at some point " in {
within(500 millis) {
ignoreMsg {
case msg: String msg != "something"
}
seqRef ! "something"
expectMsg("something")
ignoreMsg {
case msg: String msg == "1"
}
expectNoMsg
ignoreNoMsg
}
}
}
}
object TestKitUsageSpec {
// Define your test specific configuration here
val config = """
akka {
loglevel = "WARNING"
}
"""
/**
* An Actor that echoes everything you send to it
*/
class EchoActor extends Actor {
def receive = {
case msg sender ! msg
}
}
/**
* An Actor that forwards every message to a next Actor
*/
class ForwardingActor(next: ActorRef) extends Actor {
def receive = {
case msg next ! msg
}
}
/**
* An Actor that only forwards certain messages to a next Actor
*/
class FilteringActor(next: ActorRef) extends Actor {
def receive = {
case msg: String next ! msg
case _ None
}
}
/**
* An actor that sends a sequence of messages with a random head list, an
* interesting value and a random tail list. The idea is that you would
* like to test that the interesting value is received and that you cant
* be bothered with the rest
*/
class SequencingActor(next: ActorRef, head: Seq[String], tail: Seq[String])
extends Actor {
def receive = {
case msg {
head foreach { next ! _ }
next ! msg
tail foreach { next ! _ }
}
}
}
}
//#testkit-usage

View file

@ -0,0 +1,290 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.testkit
import language.postfixOps
import scala.util.Success
//#imports-test-probe
import akka.testkit.TestProbe
import scala.concurrent.util.duration._
import akka.actor._
import scala.concurrent.Future
//#imports-test-probe
import akka.testkit.AkkaSpec
import akka.testkit.DefaultTimeout
import akka.testkit.ImplicitSender
import scala.util.control.NonFatal
object TestkitDocSpec {
case object Say42
case object Unknown
class MyActor extends Actor {
def receive = {
case Say42 sender ! 42
case "some work" sender ! "some result"
}
}
//#my-double-echo
class MyDoubleEcho extends Actor {
var dest1: ActorRef = _
var dest2: ActorRef = _
def receive = {
case (d1: ActorRef, d2: ActorRef)
dest1 = d1
dest2 = d2
case x
dest1 ! x
dest2 ! x
}
}
//#my-double-echo
import akka.testkit.TestProbe
//#test-probe-forward-actors
class Source(target: ActorRef) extends Actor {
def receive = {
case "start" target ! "work"
}
}
class Destination extends Actor {
def receive = {
case x // Do something..
}
}
//#test-probe-forward-actors
class LoggingActor extends Actor {
//#logging-receive
import akka.event.LoggingReceive
def receive = LoggingReceive {
case msg // Do something...
}
//#logging-receive
}
}
class TestkitDocSpec extends AkkaSpec with DefaultTimeout with ImplicitSender {
import TestkitDocSpec._
"demonstrate usage of TestActorRef" in {
//#test-actor-ref
import akka.testkit.TestActorRef
val actorRef = TestActorRef[MyActor]
val actor = actorRef.underlyingActor
//#test-actor-ref
}
"demonstrate usage of TestFSMRef" in {
//#test-fsm-ref
import akka.testkit.TestFSMRef
import akka.actor.FSM
import scala.concurrent.util.duration._
val fsm = TestFSMRef(new Actor with FSM[Int, String] {
startWith(1, "")
when(1) {
case Event("go", _) goto(2) using "go"
}
when(2) {
case Event("back", _) goto(1) using "back"
}
})
assert(fsm.stateName == 1)
assert(fsm.stateData == "")
fsm ! "go" // being a TestActorRef, this runs also on the CallingThreadDispatcher
assert(fsm.stateName == 2)
assert(fsm.stateData == "go")
fsm.setState(stateName = 1)
assert(fsm.stateName == 1)
assert(fsm.timerActive_?("test") == false)
fsm.setTimer("test", 12, 10 millis, true)
assert(fsm.timerActive_?("test") == true)
fsm.cancelTimer("test")
assert(fsm.timerActive_?("test") == false)
//#test-fsm-ref
}
"demonstrate testing of behavior" in {
//#test-behavior
import akka.testkit.TestActorRef
import scala.concurrent.util.duration._
import scala.concurrent.Await
import akka.pattern.ask
val actorRef = TestActorRef(new MyActor)
// hypothetical message stimulating a '42' answer
val future = actorRef ? Say42
val Success(result: Int) = future.value.get
result must be(42)
//#test-behavior
}
"demonstrate unhandled message" in {
//#test-unhandled
import akka.testkit.TestActorRef
system.eventStream.subscribe(testActor, classOf[UnhandledMessage])
val ref = TestActorRef[MyActor]
ref.receive(Unknown)
expectMsg(1 second, UnhandledMessage(Unknown, system.deadLetters, ref))
//#test-unhandled
}
"demonstrate expecting exceptions" in {
//#test-expecting-exceptions
import akka.testkit.TestActorRef
val actorRef = TestActorRef(new Actor {
def receive = {
case "hello" throw new IllegalArgumentException("boom")
}
})
intercept[IllegalArgumentException] { actorRef.receive("hello") }
//#test-expecting-exceptions
}
"demonstrate within" in {
type Worker = MyActor
//#test-within
import akka.actor.Props
import scala.concurrent.util.duration._
val worker = system.actorOf(Props[Worker])
within(200 millis) {
worker ! "some work"
expectMsg("some result")
expectNoMsg // will block for the rest of the 200ms
Thread.sleep(300) // will NOT make this block fail
}
//#test-within
}
"demonstrate dilated duration" in {
//#duration-dilation
import scala.concurrent.util.duration._
import akka.testkit._
10.milliseconds.dilated
//#duration-dilation
}
"demonstrate usage of probe" in {
//#test-probe
val probe1 = TestProbe()
val probe2 = TestProbe()
val actor = system.actorOf(Props[MyDoubleEcho])
actor ! (probe1.ref, probe2.ref)
actor ! "hello"
probe1.expectMsg(500 millis, "hello")
probe2.expectMsg(500 millis, "hello")
//#test-probe
//#test-special-probe
case class Update(id: Int, value: String)
val probe = new TestProbe(system) {
def expectUpdate(x: Int) = {
expectMsgPF() {
case Update(id, _) if id == x true
}
sender ! "ACK"
}
}
//#test-special-probe
}
"demonstrate probe reply" in {
import akka.testkit.TestProbe
import scala.concurrent.util.duration._
import akka.pattern.ask
//#test-probe-reply
val probe = TestProbe()
val future = probe.ref ? "hello"
probe.expectMsg(0 millis, "hello") // TestActor runs on CallingThreadDispatcher
probe.reply("world")
assert(future.isCompleted && future.value == Some(Success("world")))
//#test-probe-reply
}
"demonstrate probe forward" in {
import akka.testkit.TestProbe
import akka.actor.Props
//#test-probe-forward
val probe = TestProbe()
val source = system.actorOf(Props(new Source(probe.ref)))
val dest = system.actorOf(Props[Destination])
source ! "start"
probe.expectMsg("work")
probe.forward(dest)
//#test-probe-forward
}
"demonstrate " in {
//#calling-thread-dispatcher
import akka.testkit.CallingThreadDispatcher
val ref = system.actorOf(Props[MyActor].withDispatcher(CallingThreadDispatcher.Id))
//#calling-thread-dispatcher
}
"demonstrate EventFilter" in {
//#event-filter
import akka.testkit.EventFilter
import com.typesafe.config.ConfigFactory
implicit val system = ActorSystem("testsystem", ConfigFactory.parseString("""
akka.event-handlers = ["akka.testkit.TestEventListener"]
"""))
try {
val actor = system.actorOf(Props.empty)
EventFilter[ActorKilledException](occurrences = 1) intercept {
actor ! Kill
}
} finally {
system.shutdown()
}
//#event-filter
}
"demonstrate TestKitBase" in {
//#test-kit-base
import akka.testkit.TestKitBase
class MyTest extends TestKitBase {
implicit lazy val system = ActorSystem()
//#put-your-test-code-here
val probe = TestProbe()
probe.send(testActor, "hello")
try expectMsg("hello") catch { case NonFatal(e) system.shutdown(); throw e }
//#put-your-test-code-here
system.shutdown()
}
//#test-kit-base
}
"demonstrate within() nesting" in {
intercept[AssertionError] {
//#test-within-probe
val probe = TestProbe()
within(1 second) {
probe.expectMsg("hello")
}
//#test-within-probe
}
}
}

View file

@ -0,0 +1,233 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.transactor
import language.postfixOps
import akka.actor._
import akka.transactor._
import scala.concurrent.util.duration._
import akka.util.Timeout
import akka.testkit._
import scala.concurrent.stm._
object CoordinatedExample {
//#coordinated-example
import akka.actor._
import akka.transactor._
import scala.concurrent.stm._
case class Increment(friend: Option[ActorRef] = None)
case object GetCount
class Counter extends Actor {
val count = Ref(0)
def receive = {
case coordinated @ Coordinated(Increment(friend)) {
friend foreach (_ ! coordinated(Increment()))
coordinated atomic { implicit t
count transform (_ + 1)
}
}
case GetCount sender ! count.single.get
}
}
//#coordinated-example
}
object CoordinatedApi {
case object Message
class Coordinator extends Actor {
//#receive-coordinated
def receive = {
case coordinated @ Coordinated(Message) {
//#coordinated-atomic
coordinated atomic { implicit t
// do something in the coordinated transaction ...
}
//#coordinated-atomic
}
}
//#receive-coordinated
}
}
object CounterExample {
//#counter-example
import akka.transactor._
import scala.concurrent.stm._
case object Increment
class Counter extends Transactor {
val count = Ref(0)
def atomically = implicit txn {
case Increment count transform (_ + 1)
}
}
//#counter-example
}
object FriendlyCounterExample {
//#friendly-counter-example
import akka.actor._
import akka.transactor._
import scala.concurrent.stm._
case object Increment
class FriendlyCounter(friend: ActorRef) extends Transactor {
val count = Ref(0)
override def coordinate = {
case Increment include(friend)
}
def atomically = implicit txn {
case Increment count transform (_ + 1)
}
}
//#friendly-counter-example
class Friend extends Transactor {
val count = Ref(0)
def atomically = implicit txn {
case Increment count transform (_ + 1)
}
}
}
// Only checked for compilation
object TransactorCoordinate {
case object Message
case object SomeMessage
case object SomeOtherMessage
case object OtherMessage
case object Message1
case object Message2
class TestCoordinateInclude(actor1: ActorRef, actor2: ActorRef, actor3: ActorRef) extends Transactor {
//#coordinate-include
override def coordinate = {
case Message include(actor1, actor2, actor3)
}
//#coordinate-include
def atomically = txn doNothing
}
class TestCoordinateSendTo(someActor: ActorRef, actor1: ActorRef, actor2: ActorRef) extends Transactor {
//#coordinate-sendto
override def coordinate = {
case SomeMessage sendTo(someActor -> SomeOtherMessage)
case OtherMessage sendTo(actor1 -> Message1, actor2 -> Message2)
}
//#coordinate-sendto
def atomically = txn doNothing
}
}
class TransactorDocSpec extends AkkaSpec {
"coordinated example" in {
import CoordinatedExample._
//#run-coordinated-example
import scala.concurrent.Await
import scala.concurrent.util.duration._
import akka.util.Timeout
import akka.pattern.ask
val system = ActorSystem("app")
val counter1 = system.actorOf(Props[Counter], name = "counter1")
val counter2 = system.actorOf(Props[Counter], name = "counter2")
implicit val timeout = Timeout(5 seconds)
counter1 ! Coordinated(Increment(Some(counter2)))
val count = Await.result(counter1 ? GetCount, timeout.duration)
// count == 1
//#run-coordinated-example
count must be === 1
system.shutdown()
}
"coordinated api" in {
import CoordinatedApi._
//#implicit-timeout
import scala.concurrent.util.duration._
import akka.util.Timeout
implicit val timeout = Timeout(5 seconds)
//#implicit-timeout
//#create-coordinated
val coordinated = Coordinated()
//#create-coordinated
val system = ActorSystem("coordinated")
val actor = system.actorOf(Props[Coordinator], name = "coordinator")
//#send-coordinated
actor ! Coordinated(Message)
//#send-coordinated
//#include-coordinated
actor ! coordinated(Message)
//#include-coordinated
coordinated.await()
system.shutdown()
}
"counter transactor" in {
import CounterExample._
val system = ActorSystem("transactors")
lazy val underlyingCounter = new Counter
val counter = system.actorOf(Props(underlyingCounter), name = "counter")
val coordinated = Coordinated()(Timeout(5 seconds))
counter ! coordinated(Increment)
coordinated.await()
underlyingCounter.count.single.get must be === 1
system.shutdown()
}
"friendly counter transactor" in {
import FriendlyCounterExample._
val system = ActorSystem("transactors")
lazy val underlyingFriend = new Friend
val friend = system.actorOf(Props(underlyingFriend), name = "friend")
lazy val underlyingFriendlyCounter = new FriendlyCounter(friend)
val friendlyCounter = system.actorOf(Props(underlyingFriendlyCounter), name = "friendly")
val coordinated = Coordinated()(Timeout(5 seconds))
friendlyCounter ! coordinated(Increment)
coordinated.await()
underlyingFriendlyCounter.count.single.get must be === 1
underlyingFriend.count.single.get must be === 1
system.shutdown()
}
}

View file

@ -0,0 +1,184 @@
/**
* Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
*/
package docs.zeromq
import language.postfixOps
import akka.actor.{ Actor, Props }
import scala.concurrent.util.duration._
import akka.testkit._
import akka.zeromq.{ ZeroMQVersion, ZeroMQExtension }
import java.text.SimpleDateFormat
import java.util.Date
import akka.zeromq.{ SocketType, Bind }
object ZeromqDocSpec {
//#health
import akka.zeromq._
import akka.actor.Actor
import akka.actor.Props
import akka.actor.ActorLogging
import akka.serialization.SerializationExtension
import java.lang.management.ManagementFactory
case object Tick
case class Heap(timestamp: Long, used: Long, max: Long)
case class Load(timestamp: Long, loadAverage: Double)
class HealthProbe extends Actor {
val pubSocket = ZeroMQExtension(context.system).newSocket(SocketType.Pub, Bind("tcp://127.0.0.1:1235"))
val memory = ManagementFactory.getMemoryMXBean
val os = ManagementFactory.getOperatingSystemMXBean
val ser = SerializationExtension(context.system)
import context.dispatcher
override def preStart() {
context.system.scheduler.schedule(1 second, 1 second, self, Tick)
}
override def postRestart(reason: Throwable) {
// don't call preStart, only schedule once
}
def receive: Receive = {
case Tick
val currentHeap = memory.getHeapMemoryUsage
val timestamp = System.currentTimeMillis
// use akka SerializationExtension to convert to bytes
val heapPayload = ser.serialize(Heap(timestamp, currentHeap.getUsed, currentHeap.getMax)).get
// the first frame is the topic, second is the message
pubSocket ! ZMQMessage(Seq(Frame("health.heap"), Frame(heapPayload)))
// use akka SerializationExtension to convert to bytes
val loadPayload = ser.serialize(Load(timestamp, os.getSystemLoadAverage)).get
// the first frame is the topic, second is the message
pubSocket ! ZMQMessage(Seq(Frame("health.load"), Frame(loadPayload)))
}
}
//#health
//#logger
class Logger extends Actor with ActorLogging {
ZeroMQExtension(context.system).newSocket(SocketType.Sub, Listener(self), Connect("tcp://127.0.0.1:1235"), Subscribe("health"))
val ser = SerializationExtension(context.system)
val timestampFormat = new SimpleDateFormat("HH:mm:ss.SSS")
def receive = {
// the first frame is the topic, second is the message
case m: ZMQMessage if m.firstFrameAsString == "health.heap"
val Heap(timestamp, used, max) = ser.deserialize(m.payload(1), classOf[Heap]).get
log.info("Used heap {} bytes, at {}", used, timestampFormat.format(new Date(timestamp)))
case m: ZMQMessage if m.firstFrameAsString == "health.load"
val Load(timestamp, loadAverage) = ser.deserialize(m.payload(1), classOf[Load]).get
log.info("Load average {}, at {}", loadAverage, timestampFormat.format(new Date(timestamp)))
}
}
//#logger
//#alerter
class HeapAlerter extends Actor with ActorLogging {
ZeroMQExtension(context.system).newSocket(SocketType.Sub, Listener(self), Connect("tcp://127.0.0.1:1235"), Subscribe("health.heap"))
val ser = SerializationExtension(context.system)
var count = 0
def receive = {
// the first frame is the topic, second is the message
case m: ZMQMessage if m.firstFrameAsString == "health.heap"
val Heap(timestamp, used, max) = ser.deserialize(m.payload(1), classOf[Heap]).get
if ((used.toDouble / max) > 0.9) count += 1
else count = 0
if (count > 10) log.warning("Need more memory, using {} %", (100.0 * used / max))
}
}
//#alerter
}
class ZeromqDocSpec extends AkkaSpec("akka.loglevel=INFO") {
import ZeromqDocSpec._
"demonstrate how to create socket" in {
checkZeroMQInstallation()
//#pub-socket
import akka.zeromq.ZeroMQExtension
val pubSocket = ZeroMQExtension(system).newSocket(SocketType.Pub, Bind("tcp://127.0.0.1:21231"))
//#pub-socket
//#sub-socket
import akka.zeromq._
val listener = system.actorOf(Props(new Actor {
def receive: Receive = {
case Connecting //...
case m: ZMQMessage //...
case _ //...
}
}))
val subSocket = ZeroMQExtension(system).newSocket(SocketType.Sub, Listener(listener), Connect("tcp://127.0.0.1:21231"), SubscribeAll)
//#sub-socket
//#sub-topic-socket
val subTopicSocket = ZeroMQExtension(system).newSocket(SocketType.Sub, Listener(listener), Connect("tcp://127.0.0.1:21231"), Subscribe("foo.bar"))
//#sub-topic-socket
//#unsub-topic-socket
subTopicSocket ! Unsubscribe("foo.bar")
//#unsub-topic-socket
val payload = Array.empty[Byte]
//#pub-topic
pubSocket ! ZMQMessage(Seq(Frame("foo.bar"), Frame(payload)))
//#pub-topic
system.stop(subSocket)
system.stop(subTopicSocket)
//#high-watermark
val highWatermarkSocket = ZeroMQExtension(system).newSocket(
SocketType.Router,
Listener(listener),
Bind("tcp://127.0.0.1:21233"),
HighWatermark(50000))
//#high-watermark
}
"demonstrate pub-sub" in {
checkZeroMQInstallation()
//#health
system.actorOf(Props[HealthProbe], name = "health")
//#health
//#logger
system.actorOf(Props[Logger], name = "logger")
//#logger
//#alerter
system.actorOf(Props[HeapAlerter], name = "alerter")
//#alerter
// Let it run for a while to see some output.
// Don't do like this in real tests, this is only doc demonstration.
Thread.sleep(3.seconds.toMillis)
}
def checkZeroMQInstallation() = try {
ZeroMQExtension(system).version match {
case ZeroMQVersion(2, 1, _) Unit
case version pending
}
} catch {
case e: LinkageError pending
}
}