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

Conflicts:
	akka-actor/src/main/resources/logback.xml
	akka-actor/src/main/scala/dataflow/DataFlowVariable.scala
	akka-actor/src/test/scala/dataflow/DataFlowSpec.scala
	akka-remote/src/main/scala/remote/RemoteClient.scala
	akka-remote/src/test/resources/logback-test.xml

akka-actor/src/main/scala/dataflow/DataFlowVariable.scala

@@ -21,8 +21,10 @@ object DataFlow {
   object Start
   object Exit
 
-  import scala.collection.JavaConversions._
+  class DataFlowVariableException(msg: String) extends AkkaException(msg)
 
+  /** Executes the supplied thunk in another thread
+   */
   def thread(body: => Unit): Unit = spawn(body)
 
   def thread[A <: AnyRef, R <: AnyRef](body: A => R) =
@@ -36,21 +38,25 @@ object DataFlow {
     }
   }
 
-  /**
-   * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
-   */
-  sealed class DataFlowVariable[T <: Any] {
-    val TIME_OUT = 1000 * 60 // 60 seconds default timeout
-
+  private object DataFlowVariable {
     private sealed abstract class DataFlowVariableMessage
     private case class Set[T <: Any](value: T) extends DataFlowVariableMessage
     private object Get extends DataFlowVariableMessage
+  }
+
+  /**
+   * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
+   */
+  sealed class DataFlowVariable[T <: Any](timeoutMs: Long) {
+    import DataFlowVariable._
+
+    def this() = this(1000 * 60)
 
     private val value = new AtomicReference[Option[T]](None)
     private val blockedReaders = new ConcurrentLinkedQueue[ActorRef]
 
     private class In[T <: Any](dataFlow: DataFlowVariable[T]) extends Actor {
-      self.timeout = TIME_OUT
+      self.timeout = timeoutMs
       def receive = {
         case s@Set(v) =>
           if (dataFlow.value.compareAndSet(None, Some(v.asInstanceOf[T]))) {
@@ -63,7 +69,7 @@ object DataFlow {
     }
 
     private class Out[T <: Any](dataFlow: DataFlowVariable[T]) extends Actor {
-      self.timeout = TIME_OUT
+      self.timeout = timeoutMs
       private var readerFuture: Option[CompletableFuture[Any]] = None
       def receive = {
         case Get => dataFlow.value.get match {
@@ -77,64 +83,42 @@ object DataFlow {
 
     private[this] val in = actorOf(new In(this)).start
 
-    def <<(ref: DataFlowVariable[T]): Unit = if (this.value.get.isEmpty) in ! Set(ref())
+    /** Sets the value of this variable (if unset) with the value of the supplied variable
+     */
+    def <<(ref: DataFlowVariable[T]) {
+      if (this.value.get.isEmpty) in ! Set(ref())
+      else throw new DataFlowVariableException(
+            "Attempt to change data flow variable (from [" + this.value.get + "] to [" + ref() + "])")
+    }
 
-    def <<(value: T): Unit = if (this.value.get.isEmpty) in ! Set(value)
+    /** Sets the value of this variable (if unset)
+     */
+    def <<(value: T) {
+      if (this.value.get.isEmpty) in ! Set(value)
+      else throw new DataFlowVariableException(
+            "Attempt to change data flow variable (from [" + this.value.get + "] to [" + value + "])")
+    }
 
+    /** Retrieves the value of variable
+     *  throws a DataFlowVariableException if it times out
+     */
     def apply(): T = {
       value.get getOrElse {
         val out = actorOf(new Out(this)).start
-        blockedReaders offer out
-        val result = (out !! Get).as[T]
-        out ! Exit
-        result.getOrElse(throw new DataFlowVariableException(
-          "Timed out (after " + TIME_OUT + " milliseconds) while waiting for result"))
+
+        val result = try {
+          blockedReaders offer out
+          (out !! Get).as[T]
+        } catch {
+          case e: Exception =>
+            out ! Exit
+            throw e
+        }
+        
+        result.getOrElse(throw new DataFlowVariableException("Timed out (after " + timeoutMs + " milliseconds) while waiting for result"))
       }
     }
 
     def shutdown = in ! Exit
   }
-
-  /**
-   * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
-   */
-  /*FIXME I do not work
-    class DataFlowStream[T <: Any] extends Seq[T] {
-    private[this] val queue = new LinkedBlockingQueue[DataFlowVariable[T]]
-
-    def <<<(ref: DataFlowVariable[T]): Boolean = queue offer ref
-
-    def <<<(value: T): Boolean = {
-      val ref = new DataFlowVariable[T]
-      ref << value
-      queue offer ref
-    }
-
-    def apply(): T = queue.take.apply
-
-    def take: DataFlowVariable[T] = queue.take
-
-    //==== For Seq ====
-
-    def length: Int = queue.size
-
-    def apply(i: Int): T = {
-      if (i == 0) apply()
-      else throw new UnsupportedOperationException(
-        "Access by index other than '0' is not supported by DataFlowStream")
-    }
-
-    def iterator: Iterator[T] = new Iterator[T] {
-      private val i = queue.iterator
-      def hasNext: Boolean = i.hasNext
-      def next: T = { val ref = i.next; ref() }
-    }
-
-    override def toList: List[T] = queue.toArray.toList.asInstanceOf[List[T]]
-  }*/
-
-  /**
-   * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
-   */
-  class DataFlowVariableException(msg: String) extends AkkaException(msg)
-}
+}