Made the repository "persistent"

README.md

@@ -89,7 +89,7 @@ To start the _betting service_ and the _betting processor_ you should create som
 > sbt start-script
 ```
 
-This will create bash scripts in `service/target/start`, `processor/target/start` and `client/target/start`. To start things you just have to run the scripts from different command prompts starting with `service` followed by `processor`.
+This will create bash scripts in `service/target/start`, `processor/target/start` and `client/target/start`. To start things you just have to run the scripts from different command prompts in the top level project directory starting with `service` followed by `processor`.
 
 If you're on Windows without access to a bash shell then you will have to copy the command line and arguments into a script of your own, and rewrite the paths to windows style paths.
 
@@ -105,6 +105,15 @@ To start testing your application you can issue the command:
 > client/target/start send
 ```
 
+And to check the results you issue the command:
+
+```
+> client/target/start
+```
+
+Remember to clean out the _persistent store_ of bets in between runs of your system. It is stored in the file `persistent_store` in the top level project directory.
+
+
 ## Authors
 
 * Henrik Engström : [@h3nk3](http://twitter.com/h3nk3)

client/src/main/scala/com/typesafe/akkademo/client/BetClient.scala

@@ -34,18 +34,18 @@ object BetClient extends App {
   val service = system.actorFor("akka://BettingServiceActorSystem@127.0.0.1:2552/user/bettingService")
 
   try {
-    if (args.size > 0 && args(0) == "send") {
-      (1 to 200).foreach {
-        p => service ! Bet("ready_player_one", p % 10 + 1, p % 100 + 1)
-      }
+    // create the list of bets
+    val bets = (1 to 200).map(p ⇒ Bet("ready_player_one", p % 10 + 1, p % 100 + 1))
+
+    if (args.size > 0 && args.head == "send") {
+      bets.foreach(bet ⇒ service ! bet)
+      println("*** SENDING OK")
     } else {
       implicit val timeout = Timeout(2 seconds)
       val fBets = ask(service, RetrieveBets).mapTo[List[Bet]]
-      Await.result(fBets, 5 seconds)
-      fBets.foreach { b => println(">> " + b) }
+      assert(Await.result(fBets, 5 seconds).sorted == bets.sorted)
+      println("*** TESTING OK")
     }
-
-    println("*** TESTING OK")
   } finally {
     system.shutdown()
   }

common/src/main/scala/com/typesafe/akkademo/common/Message.scala

@@ -12,3 +12,20 @@ case class ConfirmationMessage(id: Int)
 case object RetrieveBets
 
 case object RegisterProcessor
+
+object Bet {
+  implicit object BetOrdering extends Ordering[Bet] {
+    def compare(a: Bet, b: Bet): Int = {
+      val p = a.player compare b.player
+      if (p == 0) {
+        val g = a.game compare b.game
+        if (g == 0) {
+          a.amount compare b.amount
+        } else
+          g
+      } else
+        p
+    }
+  }
+}
+

processor/src/main/scala/com/typesafe/akkademo/processor/repository/UnstableResource.scala

@@ -4,6 +4,8 @@
 package com.typesafe.akkademo.processor.repository
 
 import com.typesafe.akkademo.common._
+import java.io.{ FileWriter, File }
+import scala.io.Source
 
 trait UnstableResource {
   def save(idempotentId: Int, player: String, game: Int, amount: Int): Unit
@@ -13,18 +15,47 @@ trait UnstableResource {
 class ReallyUnstableResource extends UnstableResource {
   val bets = scala.collection.mutable.Map[Int, Bet]()
   val randomizer = new scala.util.Random
+  val store = new File("persistent_store")
+
+  try {
+    Source.fromFile(store).getLines().foreach(s ⇒ deserialize(s).foreach {
+      case (id, player, game, amount) ⇒ if (!bets.contains(id)) bets.put(id, Bet(player, game, amount))
+    })
+  } catch {
+    case _ ⇒
+  }
 
   def save(id: Int, player: String, game: Int, amount: Int) = {
     if (id % (randomizer.nextInt(10) + 10) == 0) throw new RuntimeException("Hey, I did not count on this happening...")
     if (id % (randomizer.nextInt(17) + 17) == 0) throw new DatabaseFailureException("Help. The database's gone haywire!")
     if (id % (randomizer.nextInt(121) + 50) == 0) System.exit(1)
 
-    if (!bets.contains(id)) bets += id -> Bet(player, game, amount)
+    if (!bets.contains(id)) {
+      bets += id -> Bet(player, game, amount)
+      val fw = new FileWriter(store, true)
+      try { fw.write(serialize(id, player, game, amount) + "\n") } finally { fw.close() }
+    }
   }
 
   def findAll: Seq[Bet] = {
     bets.values.toSeq
   }
+
+  protected def serialize(id: Int, player: String, game: Int, amount: Int): String = {
+    id + ":" + player + ":" + game + ":" + amount
+  }
+
+  protected def deserialize(s: String): Option[(Int, String, Int, Int)] = {
+    s.split(":").toList match {
+      case id :: player :: game :: amount :: Nil ⇒
+        try {
+          Option((id.toInt, player, game.toInt, amount.toInt))
+        } catch {
+          case _ ⇒ None
+        }
+      case _ ⇒ None
+    }
+  }
 }
 
 class DatabaseFailureException(msg: String) extends Exception

processor/src/main/scala/com/typesafe/akkademo/processor/service/BettingProcessor.scala

@@ -16,7 +16,7 @@ class BettingProcessor extends Actor with ActorLogging {
    * Send confirmation message back to Betting service
    */
   def receive = {
-    case bet: PlayerBet =>
-    case RetrieveBets   =>
+    case bet: PlayerBet ⇒
+    case RetrieveBets   ⇒
   }
 }