diff options
Diffstat (limited to 'test')
-rw-r--r-- | test/files/jvm/future-spec/FutureTests.scala | 124 | ||||
-rw-r--r-- | test/files/jvm/future-spec/PromiseTests.scala | 15 | ||||
-rw-r--r-- | test/files/jvm/non-fatal-tests.scala | 47 | ||||
-rw-r--r-- | test/files/jvm/scala-concurrent-tck.scala | 159 |
4 files changed, 233 insertions, 112 deletions
diff --git a/test/files/jvm/future-spec/FutureTests.scala b/test/files/jvm/future-spec/FutureTests.scala index e5e01a5954..ca9ff5090f 100644 --- a/test/files/jvm/future-spec/FutureTests.scala +++ b/test/files/jvm/future-spec/FutureTests.scala @@ -10,21 +10,69 @@ import scala.runtime.NonLocalReturnControl object FutureTests extends MinimalScalaTest { - + /* some utils */ - def testAsync(s: String): Future[String] = s match { + def testAsync(s: String)(implicit ec: ExecutionContext): Future[String] = s match { case "Hello" => future { "World" } - case "Failure" => Promise.failed(new RuntimeException("Expected exception; to test fault-tolerance")).future + case "Failure" => Future.failed(new RuntimeException("Expected exception; to test fault-tolerance")) case "NoReply" => Promise[String]().future } val defaultTimeout = 5 seconds /* future specification */ + + "A future with custom ExecutionContext" should { + "shouldHandleThrowables" in { + val ms = new mutable.HashSet[Throwable] with mutable.SynchronizedSet[Throwable] + implicit val ec = scala.concurrent.ExecutionContext.fromExecutor(new scala.concurrent.forkjoin.ForkJoinPool(), { + t => + ms += t + }) + + class ThrowableTest(m: String) extends Throwable(m) + + val f1 = future[Any] { + throw new ThrowableTest("test") + } + + intercept[ThrowableTest] { + Await.result(f1, defaultTimeout) + } + + val latch = new TestLatch + val f2 = future { + Await.ready(latch, 5 seconds) + "success" + } + val f3 = f2 map { s => s.toUpperCase } + + f2 foreach { _ => throw new ThrowableTest("dispatcher foreach") } + f2 onSuccess { case _ => throw new ThrowableTest("dispatcher receive") } + + latch.open() + + Await.result(f2, defaultTimeout) mustBe ("success") + + f2 foreach { _ => throw new ThrowableTest("current thread foreach") } + f2 onSuccess { case _ => throw new ThrowableTest("current thread receive") } + + Await.result(f3, defaultTimeout) mustBe ("SUCCESS") + + val waiting = future { + Thread.sleep(1000) + } + Await.ready(waiting, 2000 millis) + + ms.size mustBe (4) + //FIXME should check + } + } - "A future" should { - + "A future with global ExecutionContext" should { + import ExecutionContext.Implicits._ + "compose with for-comprehensions" in { def async(x: Int) = future { (x * 2).toString } val future0 = future[Any] { @@ -122,20 +170,20 @@ object FutureTests extends MinimalScalaTest { val r = new IllegalStateException("recovered") intercept[IllegalStateException] { - val failed = Promise.failed[String](o).future recoverWith { - case _ if false == true => Promise.successful("yay!").future + val failed = Future.failed[String](o) recoverWith { + case _ if false == true => Future.successful("yay!") } Await.result(failed, defaultTimeout) } mustBe (o) - val recovered = Promise.failed[String](o).future recoverWith { - case _ => Promise.successful("yay!").future + val recovered = Future.failed[String](o) recoverWith { + case _ => Future.successful("yay!") } Await.result(recovered, defaultTimeout) mustBe ("yay!") intercept[IllegalStateException] { - val refailed = Promise.failed[String](o).future recoverWith { - case _ => Promise.failed[String](r).future + val refailed = Future.failed[String](o) recoverWith { + case _ => Future.failed[String](r) } Await.result(refailed, defaultTimeout) } mustBe (r) @@ -164,7 +212,7 @@ object FutureTests extends MinimalScalaTest { "firstCompletedOf" in { def futures = Vector.fill[Future[Int]](10) { Promise[Int]().future - } :+ Promise.successful[Int](5).future + } :+ Future.successful[Int](5) Await.result(Future.firstCompletedOf(futures), defaultTimeout) mustBe (5) Await.result(Future.firstCompletedOf(futures.iterator), defaultTimeout) mustBe (5) @@ -186,21 +234,21 @@ object FutureTests extends MinimalScalaTest { val timeout = 10000 millis val f = new IllegalStateException("test") intercept[IllegalStateException] { - val failed = Promise.failed[String](f).future zip Promise.successful("foo").future + val failed = Future.failed[String](f) zip Future.successful("foo") Await.result(failed, timeout) } mustBe (f) intercept[IllegalStateException] { - val failed = Promise.successful("foo").future zip Promise.failed[String](f).future + val failed = Future.successful("foo") zip Future.failed[String](f) Await.result(failed, timeout) } mustBe (f) intercept[IllegalStateException] { - val failed = Promise.failed[String](f).future zip Promise.failed[String](f).future + val failed = Future.failed[String](f) zip Future.failed[String](f) Await.result(failed, timeout) } mustBe (f) - val successful = Promise.successful("foo").future zip Promise.successful("foo").future + val successful = Future.successful("foo") zip Future.successful("foo") Await.result(successful, timeout) mustBe (("foo", "foo")) } @@ -337,50 +385,6 @@ object FutureTests extends MinimalScalaTest { Await.result(traversedIterator, defaultTimeout).sum mustBe (10000) } - "shouldHandleThrowables" in { - val ms = new mutable.HashSet[Throwable] with mutable.SynchronizedSet[Throwable] - implicit val ec = scala.concurrent.ExecutionContext.fromExecutor(new scala.concurrent.forkjoin.ForkJoinPool(), { - t => - ms += t - }) - - class ThrowableTest(m: String) extends Throwable(m) - - val f1 = future[Any] { - throw new ThrowableTest("test") - } - - intercept[ThrowableTest] { - Await.result(f1, defaultTimeout) - } - - val latch = new TestLatch - val f2 = future { - Await.ready(latch, 5 seconds) - "success" - } - val f3 = f2 map { s => s.toUpperCase } - - f2 foreach { _ => throw new ThrowableTest("dispatcher foreach") } - f2 onSuccess { case _ => throw new ThrowableTest("dispatcher receive") } - - latch.open() - - Await.result(f2, defaultTimeout) mustBe ("success") - - f2 foreach { _ => throw new ThrowableTest("current thread foreach") } - f2 onSuccess { case _ => throw new ThrowableTest("current thread receive") } - - Await.result(f3, defaultTimeout) mustBe ("SUCCESS") - - val waiting = future { - Thread.sleep(1000) - } - Await.ready(waiting, 2000 millis) - - ms.size mustBe (4) - } - "shouldBlockUntilResult" in { val latch = new TestLatch diff --git a/test/files/jvm/future-spec/PromiseTests.scala b/test/files/jvm/future-spec/PromiseTests.scala index bf9d9b39d7..49bc642b57 100644 --- a/test/files/jvm/future-spec/PromiseTests.scala +++ b/test/files/jvm/future-spec/PromiseTests.scala @@ -10,7 +10,8 @@ import scala.runtime.NonLocalReturnControl object PromiseTests extends MinimalScalaTest { - + import ExecutionContext.Implicits._ + val defaultTimeout = Inf /* promise specification */ @@ -20,11 +21,13 @@ object PromiseTests extends MinimalScalaTest { "not be completed" in { val p = Promise() p.future.isCompleted mustBe (false) + p.isCompleted mustBe (false) } "have no value" in { val p = Promise() p.future.value mustBe (None) + p.isCompleted mustBe (false) } "return supplied value on timeout" in { @@ -45,14 +48,16 @@ object PromiseTests extends MinimalScalaTest { "A successful Promise" should { val result = "test value" - val future = Promise[String]().complete(Right(result)).future - futureWithResult(_(future, result)) + val promise = Promise[String]().complete(Right(result)) + promise.isCompleted mustBe (true) + futureWithResult(_(promise.future, result)) } "A failed Promise" should { val message = "Expected Exception" - val future = Promise[String]().complete(Left(new RuntimeException(message))).future - futureWithException[RuntimeException](_(future, message)) + val promise = Promise[String]().complete(Left(new RuntimeException(message))) + promise.isCompleted mustBe (true) + futureWithException[RuntimeException](_(promise.future, message)) } "An interrupted Promise" should { diff --git a/test/files/jvm/non-fatal-tests.scala b/test/files/jvm/non-fatal-tests.scala new file mode 100644 index 0000000000..471a9d227a --- /dev/null +++ b/test/files/jvm/non-fatal-tests.scala @@ -0,0 +1,47 @@ +import scala.util.control.NonFatal + +trait NonFatalTests { + + //NonFatals + val nonFatals: Seq[Throwable] = + Seq(new StackOverflowError, + new RuntimeException, + new Exception, + new Throwable) + + //Fatals + val fatals: Seq[Throwable] = + Seq(new InterruptedException, + new OutOfMemoryError, + new LinkageError, + new VirtualMachineError {}, + new Throwable with scala.util.control.ControlThrowable, + new NotImplementedError) + + def testFatalsUsingApply(): Unit = { + fatals foreach { t => assert(NonFatal(t) == false) } + } + + def testNonFatalsUsingApply(): Unit = { + nonFatals foreach { t => assert(NonFatal(t) == true) } + } + + def testFatalsUsingUnapply(): Unit = { + fatals foreach { t => assert(NonFatal.unapply(t).isEmpty) } + } + + def testNonFatalsUsingUnapply(): Unit = { + nonFatals foreach { t => assert(NonFatal.unapply(t).isDefined) } + } + + testFatalsUsingApply() + testNonFatalsUsingApply() + testFatalsUsingUnapply() + testNonFatalsUsingUnapply() +} + +object Test +extends App +with NonFatalTests { + System.exit(0) +}
\ No newline at end of file diff --git a/test/files/jvm/scala-concurrent-tck.scala b/test/files/jvm/scala-concurrent-tck.scala index 407027f904..5c9c71f3f8 100644 --- a/test/files/jvm/scala-concurrent-tck.scala +++ b/test/files/jvm/scala-concurrent-tck.scala @@ -3,22 +3,19 @@ import scala.concurrent.{ Promise, TimeoutException, SyncVar, - ExecutionException + ExecutionException, + ExecutionContext } -import scala.concurrent.future -import scala.concurrent.promise -import scala.concurrent.blocking +import scala.concurrent.{ future, promise, blocking } import scala.util.{ Try, Success, Failure } - import scala.concurrent.util.Duration - trait TestBase { def once(body: (() => Unit) => Unit) { val sv = new SyncVar[Boolean] body(() => sv put true) - sv.take() + sv.take(2000) } // def assert(cond: => Boolean) { @@ -33,7 +30,8 @@ trait TestBase { trait FutureCallbacks extends TestBase { - + import ExecutionContext.Implicits._ + def testOnSuccess(): Unit = once { done => var x = 0 @@ -147,6 +145,7 @@ trait FutureCallbacks extends TestBase { trait FutureCombinators extends TestBase { + import ExecutionContext.Implicits._ def testMapSuccess(): Unit = once { done => @@ -591,7 +590,8 @@ trait FutureCombinators extends TestBase { trait FutureProjections extends TestBase { - + import ExecutionContext.Implicits._ + def testFailedFailureOnComplete(): Unit = once { done => val cause = new RuntimeException @@ -673,7 +673,8 @@ trait FutureProjections extends TestBase { trait Blocking extends TestBase { - + import ExecutionContext.Implicits._ + def testAwaitSuccess(): Unit = once { done => val f = future { 0 } @@ -702,8 +703,67 @@ trait Blocking extends TestBase { } +trait BlockContexts extends TestBase { + import ExecutionContext.Implicits._ + import scala.concurrent.{ Await, Awaitable, BlockContext } + + private def getBlockContext(body: => BlockContext): BlockContext = { + blocking(Future { body }, Duration(500, "ms")) + } + + // test outside of an ExecutionContext + def testDefaultOutsideFuture(): Unit = { + val bc = BlockContext.current + assert(bc.getClass.getName.contains("DefaultBlockContext")) + } + + // test BlockContext in our default ExecutionContext + def testDefaultFJP(): Unit = { + val bc = getBlockContext(BlockContext.current) + assert(bc.isInstanceOf[scala.concurrent.forkjoin.ForkJoinWorkerThread]) + } + + // test BlockContext inside BlockContext.withBlockContext + def testPushCustom(): Unit = { + val orig = BlockContext.current + val customBC = new BlockContext() { + override def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T = + orig.internalBlockingCall(awaitable, atMost) + } + + val bc = getBlockContext({ + BlockContext.withBlockContext(customBC) { + BlockContext.current + } + }) + + assert(bc eq customBC) + } + + // test BlockContext after a BlockContext.push + def testPopCustom(): Unit = { + val orig = BlockContext.current + val customBC = new BlockContext() { + override def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T = + orig.internalBlockingCall(awaitable, atMost) + } + + val bc = getBlockContext({ + BlockContext.withBlockContext(customBC) {} + BlockContext.current + }) + + assert(bc ne customBC) + } + + testDefaultOutsideFuture() + testDefaultFJP() + testPushCustom() + testPopCustom() +} trait Promises extends TestBase { + import ExecutionContext.Implicits._ def testSuccess(): Unit = once { done => @@ -730,7 +790,8 @@ trait Promises extends TestBase { trait Exceptions extends TestBase { - + import ExecutionContext.Implicits._ + } // trait TryEitherExtractor extends TestBase { @@ -811,7 +872,7 @@ trait Exceptions extends TestBase { trait CustomExecutionContext extends TestBase { import scala.concurrent.{ ExecutionContext, Awaitable } - def defaultEC = ExecutionContext.defaultExecutionContext + def defaultEC = ExecutionContext.global val inEC = new java.lang.ThreadLocal[Int]() { override def initialValue = 0 @@ -826,7 +887,7 @@ trait CustomExecutionContext extends TestBase { val _count = new java.util.concurrent.atomic.AtomicInteger(0) def count = _count.get - def delegate = ExecutionContext.defaultExecutionContext + def delegate = ExecutionContext.global override def execute(runnable: Runnable) = { _count.incrementAndGet() @@ -843,9 +904,6 @@ trait CustomExecutionContext extends TestBase { delegate.execute(wrapper) } - override def internalBlockingCall[T](awaitable: Awaitable[T], atMost: Duration): T = - delegate.internalBlockingCall(awaitable, atMost) - override def reportFailure(t: Throwable): Unit = { System.err.println("Failure: " + t.getClass.getSimpleName + ": " + t.getMessage) delegate.reportFailure(t) @@ -860,14 +918,16 @@ trait CustomExecutionContext extends TestBase { def testOnSuccessCustomEC(): Unit = { val count = countExecs { implicit ec => - once { done => - val f = future({ assertNoEC() })(defaultEC) - f onSuccess { - case _ => - assertEC() + blocking { + once { done => + val f = future({ assertNoEC() })(defaultEC) + f onSuccess { + case _ => + assertEC() done() + } + assertNoEC() } - assertNoEC() } } @@ -877,12 +937,14 @@ trait CustomExecutionContext extends TestBase { def testKeptPromiseCustomEC(): Unit = { val count = countExecs { implicit ec => - once { done => - val f = Promise.successful(10).future - f onSuccess { - case _ => - assertEC() + blocking { + once { done => + val f = Promise.successful(10).future + f onSuccess { + case _ => + assertEC() done() + } } } } @@ -893,28 +955,30 @@ trait CustomExecutionContext extends TestBase { def testCallbackChainCustomEC(): Unit = { val count = countExecs { implicit ec => - once { done => - assertNoEC() - val addOne = { x: Int => assertEC(); x + 1 } - val f = Promise.successful(10).future - f.map(addOne).filter { x => - assertEC() - x == 11 - } flatMap { x => - Promise.successful(x + 1).future.map(addOne).map(addOne) - } onComplete { - case Left(t) => - try { - throw new AssertionError("error in test: " + t.getMessage, t) - } finally { + blocking { + once { done => + assertNoEC() + val addOne = { x: Int => assertEC(); x + 1 } + val f = Promise.successful(10).future + f.map(addOne).filter { x => + assertEC() + x == 11 + } flatMap { x => + Promise.successful(x + 1).future.map(addOne).map(addOne) + } onComplete { + case Left(t) => + try { + throw new AssertionError("error in test: " + t.getMessage, t) + } finally { + done() + } + case Right(x) => + assertEC() + assert(x == 14) done() - } - case Right(x) => - assertEC() - assert(x == 14) - done() + } + assertNoEC() } - assertNoEC() } } @@ -934,6 +998,7 @@ with FutureCallbacks with FutureCombinators with FutureProjections with Promises +with BlockContexts with Exceptions // with TryEitherExtractor with CustomExecutionContext |