Merge branch 'master' into wip-fix-multi-jvm-test-jboner

akka-actor-tests/src/test/java/akka/dispatch/JavaFutureTests.java

@@ -306,10 +306,10 @@ public Object recover(Throwable t) throws Throwable {
   }
 
   @Test
-  public void tryRecoverToMustBeCallable() {
+  public void recoverWithToMustBeCallable() {
     final IllegalStateException fail = new IllegalStateException("OHNOES");
     Promise<Object> p = Futures.promise(system.dispatcher());
-    Future<Object> f = p.future().tryRecover(new Recover<Future<Object>>() {
+    Future<Object> f = p.future().recoverWith(new Recover<Future<Object>>() {
       public Future<Object> recover(Throwable t) throws Throwable {
         if (t == fail) return Futures.<Object>successful("foo", system.dispatcher()).future();
         else throw t;

akka-actor-tests/src/test/scala/akka/actor/ActorTimeoutSpec.scala

@@ -13,20 +13,21 @@ import akka.pattern.{ ask, AskTimeoutException }
 class ActorTimeoutSpec extends AkkaSpec {
 
   val testTimeout = 200.millis.dilated
+  val leeway = 500.millis.dilated
 
   "An Actor-based Future" must {
 
     "use implicitly supplied timeout" in {
       implicit val timeout = Timeout(testTimeout)
       val echo = system.actorOf(Props.empty)
       val f = (echo ? "hallo")
-      intercept[AskTimeoutException] { Await.result(f, testTimeout * 2) }
+      intercept[AskTimeoutException] { Await.result(f, testTimeout + leeway) }
     }
 
     "use explicitly supplied timeout" in {
       val echo = system.actorOf(Props.empty)
       val f = echo.?("hallo")(testTimeout)
-      intercept[AskTimeoutException] { Await.result(f, testTimeout * 2) }
+      intercept[AskTimeoutException] { Await.result(f, testTimeout + leeway) }
     }
   }
 }

akka-actor-tests/src/test/scala/akka/actor/DeployerSpec.scala

@@ -64,7 +64,7 @@ class DeployerSpec extends AkkaSpec(DeployerSpec.deployerConf) {
           service,
           deployment.get.config,
           NoRouter,
-          LocalScope)))
+          NoScopeGiven)))
     }
 
     "use None deployment for undefined service" in {
@@ -117,9 +117,9 @@ class DeployerSpec extends AkkaSpec(DeployerSpec.deployerConf) {
       val deployment = system.asInstanceOf[ActorSystemImpl].provider.deployer.lookup(service)
       deployment must be('defined)
       deployment.get.path must be(service)
-      deployment.get.routing.getClass must be(expected.getClass)
-      deployment.get.routing.resizer must be(expected.resizer)
-      deployment.get.scope must be(LocalScope)
+      deployment.get.routerConfig.getClass must be(expected.getClass)
+      deployment.get.routerConfig.resizer must be(expected.resizer)
+      deployment.get.scope must be(NoScopeGiven)
     }
 
   }

akka-actor-tests/src/test/scala/akka/actor/IOActor.scala

@@ -4,7 +4,7 @@
 
 package akka.actor
 
-import akka.util.{ ByteString, Duration, Timer }
+import akka.util.{ ByteString, Duration, Deadline }
 import akka.util.duration._
 import scala.util.continuations._
 import akka.testkit._
@@ -244,13 +244,13 @@ class IOActorSpec extends AkkaSpec with DefaultTimeout {
 
     val promise = Promise[T]()(executor)
 
-    val timer = timeout match {
-      case Some(duration) ⇒ Some(Timer(duration))
+    val timer: Option[Deadline] = timeout match {
+      case Some(duration) ⇒ Some(duration fromNow)
       case None           ⇒ None
     }
 
     def check(n: Int, e: Throwable): Boolean =
-      (count.isEmpty || (n < count.get)) && (timer.isEmpty || timer.get.isTicking) && (filter.isEmpty || filter.get(e))
+      (count.isEmpty || (n < count.get)) && (timer.isEmpty || timer.get.hasTimeLeft()) && (filter.isEmpty || filter.get(e))
 
     def run(n: Int) {
       future onComplete {

akka-actor-tests/src/test/scala/akka/actor/SchedulerSpec.scala

@@ -33,7 +33,7 @@ class SchedulerSpec extends AkkaSpec with BeforeAndAfterEach with DefaultTimeout
       collectCancellable(system.scheduler.schedule(0 milliseconds, 50 milliseconds, tickActor, Tick))
 
       // after max 1 second it should be executed at least the 3 times already
-      assert(countDownLatch.await(1, TimeUnit.SECONDS))
+      assert(countDownLatch.await(2, TimeUnit.SECONDS))
 
       val countDownLatch2 = new CountDownLatch(3)
 
@@ -43,14 +43,21 @@ class SchedulerSpec extends AkkaSpec with BeforeAndAfterEach with DefaultTimeout
       assert(countDownLatch2.await(2, TimeUnit.SECONDS))
     }
 
-    "should stop continuous scheduling if the receiving actor has been terminated" in {
+    "should stop continuous scheduling if the receiving actor has been terminated" taggedAs TimingTest in {
+      val actor = system.actorOf(Props(new Actor {
+        def receive = {
+          case x ⇒ testActor ! x
+        }
+      }))
+
       // run immediately and then every 100 milliseconds
-      collectCancellable(system.scheduler.schedule(0 milliseconds, 100 milliseconds, testActor, "msg"))
+      collectCancellable(system.scheduler.schedule(0 milliseconds, 100 milliseconds, actor, "msg"))
+      expectNoMsg(1 second)
 
       // stop the actor and, hence, the continuous messaging from happening
-      testActor ! PoisonPill
+      actor ! PoisonPill
 
-      expectNoMsg(500 milliseconds)
+      expectMsg("msg")
     }
 
     "schedule once" in {
@@ -69,7 +76,7 @@ class SchedulerSpec extends AkkaSpec with BeforeAndAfterEach with DefaultTimeout
       countDownLatch.getCount must be(3)
 
       // after 1 second the wait should fail
-      assert(countDownLatch.await(1, TimeUnit.SECONDS) == false)
+      assert(countDownLatch.await(2, TimeUnit.SECONDS) == false)
       // should still be 1 left
       countDownLatch.getCount must be(1)
     }
@@ -93,7 +100,7 @@ class SchedulerSpec extends AkkaSpec with BeforeAndAfterEach with DefaultTimeout
       assert(ticks.await(3, TimeUnit.SECONDS) == false) //No counting down should've been made
     }
 
-    "be cancellable during initial delay" in {
+    "be cancellable during initial delay" taggedAs TimingTest in {
       val ticks = new AtomicInteger
 
       val initialDelay = 200.milliseconds.dilated
@@ -108,7 +115,7 @@ class SchedulerSpec extends AkkaSpec with BeforeAndAfterEach with DefaultTimeout
       ticks.get must be(0)
     }
 
-    "be cancellable after initial delay" in {
+    "be cancellable after initial delay" taggedAs TimingTest in {
       val ticks = new AtomicInteger
 
       val initialDelay = 20.milliseconds.dilated
@@ -179,7 +186,7 @@ class SchedulerSpec extends AkkaSpec with BeforeAndAfterEach with DefaultTimeout
       Await.ready(ticks, 3 seconds)
     }
 
-    "schedule with different initial delay and frequency" in {
+    "schedule with different initial delay and frequency" taggedAs TimingTest in {
       val ticks = new TestLatch(3)
 
       case object Msg

akka-actor-tests/src/test/scala/akka/dataflow/Future2Actor.scala

@@ -5,11 +5,10 @@ package akka.dataflow
 
 import akka.actor.{ Actor, Props }
 import akka.dispatch.{ Future, Await }
-import akka.actor.future2actor
 import akka.util.duration._
 import akka.testkit.AkkaSpec
 import akka.testkit.DefaultTimeout
-import akka.pattern.ask
+import akka.pattern.{ ask, pipe }
 
 class Future2ActorSpec extends AkkaSpec with DefaultTimeout {
 

akka-actor-tests/src/test/scala/akka/dispatch/FutureSpec.scala

@@ -302,18 +302,18 @@ class FutureSpec extends AkkaSpec with Checkers with BeforeAndAfterAll with Defa
         }
       }
 
-      "tryRecover from exceptions" in {
+      "recoverWith from exceptions" in {
         val o = new IllegalStateException("original")
         val r = new IllegalStateException("recovered")
 
         intercept[IllegalStateException] {
-          Await.result(Promise.failed[String](o) tryRecover { case _ if false == true ⇒ Promise.successful("yay!") }, timeout.duration)
+          Await.result(Promise.failed[String](o) recoverWith { case _ if false == true ⇒ Promise.successful("yay!") }, timeout.duration)
         } must be(o)
 
-        Await.result(Promise.failed[String](o) tryRecover { case _ ⇒ Promise.successful("yay!") }, timeout.duration) must equal("yay!")
+        Await.result(Promise.failed[String](o) recoverWith { case _ ⇒ Promise.successful("yay!") }, timeout.duration) must equal("yay!")
 
         intercept[IllegalStateException] {
-          Await.result(Promise.failed[String](o) tryRecover { case _ ⇒ Promise.failed[String](r) }, timeout.duration)
+          Await.result(Promise.failed[String](o) recoverWith { case _ ⇒ Promise.failed[String](r) }, timeout.duration)
         } must be(r)
       }
 

akka-actor-tests/src/test/scala/akka/routing/ConfiguredLocalRoutingSpec.scala

@@ -1,12 +1,19 @@
+/**
+ * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
+ */
 package akka.routing
 
-import akka.actor._
-import akka.routing._
 import java.util.concurrent.atomic.AtomicInteger
-import akka.testkit._
-import akka.util.duration._
+
+import org.junit.runner.RunWith
+
+import akka.actor.actorRef2Scala
+import akka.actor.{ Props, LocalActorRef, Deploy, Actor }
+import akka.config.ConfigurationException
 import akka.dispatch.Await
 import akka.pattern.ask
+import akka.testkit.{ TestLatch, ImplicitSender, DefaultTimeout, AkkaSpec }
+import akka.util.duration.intToDurationInt
 
 object ConfiguredLocalRoutingSpec {
   val config = """
@@ -19,6 +26,12 @@ object ConfiguredLocalRoutingSpec {
             core-pool-size-max = 16
           }
         }
+        deployment {
+          /config {
+            router = random
+            nr-of-instances = 4
+          }
+        }
       }
     }
   """
@@ -27,18 +40,52 @@ object ConfiguredLocalRoutingSpec {
 @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
 class ConfiguredLocalRoutingSpec extends AkkaSpec(ConfiguredLocalRoutingSpec.config) with DefaultTimeout with ImplicitSender {
 
-  val deployer = system.asInstanceOf[ActorSystemImpl].provider.deployer
-
   "RouterConfig" must {
 
+    "be picked up from Props" in {
+      val actor = system.actorOf(Props(new Actor {
+        def receive = {
+          case "get" ⇒ sender ! context.props
+        }
+      }).withRouter(RoundRobinRouter(12)), "someOther")
+      actor.asInstanceOf[LocalActorRef].underlying.props.routerConfig must be === RoundRobinRouter(12)
+      system.stop(actor)
+    }
+
     "be overridable in config" in {
-      deployer.deploy(Deploy("/config", null, RandomRouter(4), LocalScope))
       val actor = system.actorOf(Props(new Actor {
         def receive = {
           case "get" ⇒ sender ! context.props
         }
       }).withRouter(RoundRobinRouter(12)), "config")
       actor.asInstanceOf[LocalActorRef].underlying.props.routerConfig must be === RandomRouter(4)
+      system.stop(actor)
+    }
+
+    "be overridable in explicit deployment" in {
+      val actor = system.actorOf(Props(new Actor {
+        def receive = {
+          case "get" ⇒ sender ! context.props
+        }
+      }).withRouter(FromConfig).withDeploy(Deploy(routerConfig = RoundRobinRouter(12))), "someOther")
+      actor.asInstanceOf[LocalActorRef].underlying.props.routerConfig must be === RoundRobinRouter(12)
+      system.stop(actor)
+    }
+
+    "be overridable in config even with explicit deployment" in {
+      val actor = system.actorOf(Props(new Actor {
+        def receive = {
+          case "get" ⇒ sender ! context.props
+        }
+      }).withRouter(FromConfig).withDeploy(Deploy(routerConfig = RoundRobinRouter(12))), "config")
+      actor.asInstanceOf[LocalActorRef].underlying.props.routerConfig must be === RandomRouter(4)
+      system.stop(actor)
+    }
+
+    "fail with an exception if not correct" in {
+      intercept[ConfigurationException] {
+        system.actorOf(Props.empty.withRouter(FromConfig))
+      }
     }
 
   }

akka-actor-tests/src/test/scala/akka/util/NonFatalSpec.scala

@@ -0,0 +1,64 @@
+/**
+ * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
+ */
+package akka.util
+
+import org.scalatest.matchers.MustMatchers
+import akka.testkit.AkkaSpec
+
+@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
+class NonFatalSpec extends AkkaSpec with MustMatchers {
+
+  "A NonFatal extractor" must {
+
+    "match ordinary RuntimeException" in {
+      try {
+        throw new RuntimeException("Boom")
+      } catch {
+        case NonFatal(e) ⇒ // as expected
+      }
+    }
+
+    "not match StackOverflowError" in {
+      //not @tailrec
+      def blowUp(n: Long): Long = {
+        blowUp(n + 1) + 1
+      }
+
+      intercept[StackOverflowError] {
+        try {
+          blowUp(0)
+        } catch {
+          case NonFatal(e) ⇒ assert(false)
+        }
+      }
+    }
+
+    "not match InterruptedException" in {
+      intercept[InterruptedException] {
+        try {
+          throw new InterruptedException("Simulated InterruptedException")
+        } catch {
+          case NonFatal(e) ⇒ assert(false)
+        }
+      }
+    }
+
+    "be used together with InterruptedException" in {
+      try {
+        throw new InterruptedException("Simulated InterruptedException")
+      } catch {
+        case _: InterruptedException ⇒ // as expected
+        case NonFatal(e)             ⇒ assert(false)
+      }
+
+      try {
+        throw new RuntimeException("Simulated RuntimeException")
+      } catch {
+        case NonFatal(_) | _: InterruptedException ⇒ // as expected
+      }
+    }
+
+  }
+
+}