Fast Track to Akka (Sep 26-27): Training Materials

gistfile1.txt

Slides:
https://www.dropbox.com/s/5fjq0kdpb25h57u/fast-track-to-akka-preparations.pdf
https://www.dropbox.com/s/wlqz8o3pcndkx4q/fast-track-to-akka.pdf
https://www.dropbox.com/s/1p4mdd100aw3bkz/fast-track-to-akka-part2.pdf

Start project:
https://www.dropbox.com/s/96pwbnt19r0lcni/akkatrain.zip

NEW! Solutions:
https://www.dropbox.com/s/42o8lm3khbgads0/training-akka-solutions.zip

Akka Cluster Specification:
http://doc.akka.io/docs/akka/snapshot/cluster/cluster.html

NEW! Typesafe Console Announcement:
http://blog.typesafe.com/typesafe-console-becomes-part-of-the-stack-87576


Template for testing:
=====================

package akka.training

import akka.testkit.{ TestKit, ImplicitSender }
import akka.actor.{ ActorSystem, Props }
import org.specs2.mutable.Specification
import org.specs2.time.{ NoTimeConversions => NTC }
import org.specs2.runner.JUnitRunner

@org.junit.runner.RunWith(classOf[JUnitRunner])
class ActorSpec extends TestKit(ActorSystem()) with
      ImplicitSender with Specification with NTC {
  
  "A ComputeActor" should {
    "respond with the length of a string" in {
      val ref = system.actorOf(Props[ComputeActor])
      ref ! "Hello world"
      expectMsg(11)
      done // necessary because Specs2 wants a matcher
    }
  }
}


QueryActor using ask and pipeTo:
================================

import akka.actor.{ Actor, ActorRef }
import akka.pattern.{ ask, pipe }
import akka.util.{ Timeout, Duration }

case class Query(s: String)
case class DbRequest(s: String)
case class Response(s: String)

class QueryActor(dbActor: ActorRef) extends Actor {
  implicit val to: Timeout = Duration("10 seconds")
  
  def receive = {
    case Query(s) =>
      println("received query "+s)
      //dbActor forward DbRequest(s)
      
      (dbActor ? DbRequest(s)) map {
        respFromDb => Response(respFromDb.toString)
      } pipeTo sender
  }
  
}


Canceling Futures Example:
==========================

import akka.dispatch.{ Promise, Future, ExecutionContext, Await }
import akka.actor.ActorSystem
import akka.util.Duration
import scala.concurrent.SyncVar
import java.util.concurrent.TimeoutException

object CancellingFutures extends App {

  val system = ActorSystem() // create an ActorSystem
  implicit val ec: ExecutionContext = system.dispatcher

  def expensiveCalc(): String = {
    try {
      val res = "hello" + 42
      Thread.sleep(2000)
      res + "afterwards"
    } catch {
      case ie: InterruptedException =>
        println("ok, cool")
        throw ie
    }
  }
  
  def cancellableFuture: Future[String] = {
    val p = Promise[String]()
    
    val theThread = new SyncVar[Thread]
    
    Future {
      theThread.put(Thread.currentThread) 
      val res = expensiveCalc()
      p.tryComplete(Right(res))
    }
    
    system.scheduler.scheduleOnce(Duration("1 seconds")){
      if (p.tryComplete(Left(new TimeoutException))) {
        // cancel computation to free up worker which is sleeping
        theThread.get.interrupt()
      }
    }

    p.future
  }
  
  val fut = cancellableFuture
  Await.ready(fut, Duration("10 seconds"))
  system.shutdown()
}