Merge branch 'master' of github.com:jboner/akka

akka-stm/src/main/scala/akka/agent/Agent.scala

@@ -7,12 +7,13 @@ package akka.agent
 import akka.stm._
 import akka.actor.Actor
 import akka.japi.{Function => JFunc, Procedure => JProc}
-import akka.dispatch.Dispatchers
+import akka.dispatch.{Dispatchers, Future}
 
 /**
  * Used internally to send functions.
  */
 private[akka] case class Update[T](function: T => T)
+private[akka] case object Get
 
 /**
  * Factory method for creating an Agent.
@@ -139,6 +140,17 @@ class Agent[T](initialValue: T) {
     value
   })
 
+  /**
+   * A future to the current value that will be completed after any currently
+   * queued updates.
+   */
+  def future(): Future[T] = (updater !!! Get).asInstanceOf[Future[T]]
+
+  /**
+   * Gets this agent's value after all currently queued updates have completed.
+   */
+  def await(): T = future.await.result.get
+
   /**
    * Map this agent to a new agent, applying the function to the internal state.
    * Does not change the value of this agent.
@@ -221,6 +233,7 @@ class AgentUpdater[T](agent: Agent[T]) extends Actor {
   def receive = {
     case update: Update[T] =>
       atomic(txFactory) { agent.ref alter update.function }
+    case Get => self reply agent.get
     case _ => ()
   }
 }

akka-stm/src/test/scala/agent/AgentSpec.scala

@@ -113,6 +113,28 @@ class AgentSpec extends WordSpec with MustMatchers {
       agent.close
     }
 
+    "be able to return a 'queued' future" in {
+      val agent = Agent("a")
+      agent send (_ + "b")
+      agent send (_ + "c")
+
+      val future = agent.future
+
+      future.await.result.get must be ("abc")
+
+      agent.close
+    }
+
+    "be able to await the value after updates have completed" in {
+      val agent = Agent("a")
+      agent send (_ + "b")
+      agent send (_ + "c")
+
+      agent.await must be ("abc")
+
+      agent.close
+    }
+
     "be able to be mapped" in {
       val agent1 = Agent(5)
       val agent2 = agent1 map (_ * 2)