Don't return Iterable(null) when Probe.fishForMessage times out (ak…

…ka#25629) * Fix for issue akka#25616 I haven't had a chance to test this, but it looks to me that this is the solution to issue akka#25616. My stacktrace in this issue is this: ``` java.lang.NullPointerException at eu.phy_awaas.irods.typed.lockingsystem.IRODSLockManagerSpec.$anonfun$new$20(IRODSLockManagerSpec.scala:91) at akka.actor.testkit.typed.internal.TestProbeImpl.loop$1(TestProbeImpl.scala:189) at akka.actor.testkit.typed.internal.TestProbeImpl.loop$1(TestProbeImpl.scala:201) at akka.actor.testkit.typed.internal.TestProbeImpl.fishForMessage_internal(TestProbeImpl.scala:215) at akka.actor.testkit.typed.internal.TestProbeImpl.fishForMessage_internal(TestProbeImpl.scala:45) at akka.actor.testkit.typed.scaladsl.TestProbe.fishForMessage(TestProbe.scala:175) at akka.actor.testkit.typed.scaladsl.TestProbe.fishForMessage$(TestProbe.scala:174) at akka.actor.testkit.typed.internal.TestProbeImpl.fishForMessage(TestProbeImpl.scala:45) at akka.actor.testkit.typed.scaladsl.TestProbe.fishForMessage(TestProbe.scala:181) at akka.actor.testkit.typed.scaladsl.TestProbe.fishForMessage$(TestProbe.scala:180) at akka.actor.testkit.typed.internal.TestProbeImpl.fishForMessage(TestProbeImpl.scala:45) at eu.phy_awaas.irods.typed.lockingsystem.IRODSLockManagerSpec.$anonfun$new$18(IRODSLockManagerSpec.scala:88) at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85) at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83) at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104) at org.scalatest.Transformer.apply(Transformer.scala:22) at org.scalatest.Transformer.apply(Transformer.scala:20) at org.scalatest.WordSpecLike$$anon$1.apply(WordSpecLike.scala:1078) at org.scalatest.TestSuite.withFixture(TestSuite.scala:196) at org.scalatest.TestSuite.withFixture$(TestSuite.scala:195) at org.scalatest.WordSpec.withFixture(WordSpec.scala:1881) at org.scalatest.WordSpecLike.invokeWithFixture$1(WordSpecLike.scala:1076) at org.scalatest.WordSpecLike.$anonfun$runTest$1(WordSpecLike.scala:1088) at org.scalatest.SuperEngine.runTestImpl(Engine.scala:289) at org.scalatest.WordSpecLike.runTest(WordSpecLike.scala:1088) at org.scalatest.WordSpecLike.runTest$(WordSpecLike.scala:1070) at org.scalatest.WordSpec.runTest(WordSpec.scala:1881) at org.scalatest.WordSpecLike.$anonfun$runTests$1(WordSpecLike.scala:1147) at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:396) at scala.collection.immutable.List.foreach(List.scala:389) at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384) at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:373) at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:410) at scala.collection.immutable.List.foreach(List.scala:389) at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384) at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:373) at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:410) at scala.collection.immutable.List.foreach(List.scala:389) at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384) at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:379) at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:461) at org.scalatest.WordSpecLike.runTests(WordSpecLike.scala:1147) at org.scalatest.WordSpecLike.runTests$(WordSpecLike.scala:1146) at org.scalatest.WordSpec.runTests(WordSpec.scala:1881) at org.scalatest.Suite.run(Suite.scala:1147) at org.scalatest.Suite.run$(Suite.scala:1129) at org.scalatest.WordSpec.org$scalatest$WordSpecLike$$super$run(WordSpec.scala:1881) at org.scalatest.WordSpecLike.$anonfun$run$1(WordSpecLike.scala:1192) at org.scalatest.SuperEngine.runImpl(Engine.scala:521) at org.scalatest.WordSpecLike.run(WordSpecLike.scala:1192) at org.scalatest.WordSpecLike.run$(WordSpecLike.scala:1190) at eu.phy_awaas.irods.typed.lockingsystem.IRODSLockManagerSpec.org$scalatest$BeforeAndAfterAll$$super$run(IRODSLockManagerSpec.scala:22) at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:213) at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210) at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208) at eu.phy_awaas.irods.typed.lockingsystem.IRODSLockManagerSpec.run(IRODSLockManagerSpec.scala:22) at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:45) at org.scalatest.tools.Runner$.$anonfun$doRunRunRunDaDoRunRun$13(Runner.scala:1340) at org.scalatest.tools.Runner$.$anonfun$doRunRunRunDaDoRunRun$13$adapted(Runner.scala:1334) at scala.collection.immutable.List.foreach(List.scala:389) at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:1334) at org.scalatest.tools.Runner$.$anonfun$runOptionallyWithPassFailReporter$24(Runner.scala:1031) at org.scalatest.tools.Runner$.$anonfun$runOptionallyWithPassFailReporter$24$adapted(Runner.scala:1010) at org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:1500) at org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:1010) at org.scalatest.tools.Runner$.run(Runner.scala:850) at org.scalatest.tools.Runner.run(Runner.scala) at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.runScalaTest2(ScalaTestRunner.java:131) at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.main(ScalaTestRunner.java:28) ``` With the calling code looking like this: ```scala val lockResults = probe.fishForMessage(Duration.create(10000, "ms"))(l => { r += 1 if (r < 5) { l.lock.use(())//crash happens here FishingOutcomes.continue } else { l.lock.use(()) FishingOutcomes.complete } }) ``` It's pretty obvious receiveOne(timeout) returns a null when it receives nothing before the timeout, passes that along to my fisher function, which tries to perform work on the supposed message only to get a null instead. * Added missing brace * Manually formatted my code the way akka likes I don't have the set up to compile this right now, so I just manually inserted the formatting that jenkins mentioned. * More changes to fit the akka formatting standard * Test for the new assertion error when fishing times out. * Cleaned up a flaw I added from when I initially put this in async spec
dwijnand · Oct 15, 2018 · f32d6c5 · f32d6c5
1 parent af240d7
commit f32d6c5
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 25 deletions.
diff --git a/...-actor-testkit-typed/src/main/scala/akka/actor/testkit/typed/internal/TestProbeImpl.scala b/...-actor-testkit-typed/src/main/scala/akka/actor/testkit/typed/internal/TestProbeImpl.scala
@@ -184,31 +184,37 @@ private[akka] final class TestProbeImpl[M](name: String, system: ActorSystem[_])
     // not tailrec but that should be ok
     def loop(timeout: FiniteDuration, seen: List[M]): List[M] = {
       val start = System.nanoTime()
-      val msg = receiveOne(timeout)
-      try {
-        fisher(msg) match {
-          case FishingOutcome.Complete    ⇒ (msg :: seen).reverse
-          case FishingOutcome.Fail(error) ⇒ throw new AssertionError(s"$error, hint: $hint")
-          case continue ⇒
-            val newTimeout =
-              if (timeout.isFinite()) timeout - (System.nanoTime() - start).nanos
-              else timeout
-            if (newTimeout.toMillis <= 0) {
-              throw new AssertionError(s"timeout ($max) during fishForMessage, seen messages ${seen.reverse}, hint: $hint")
-            } else {
-
-              continue match {
-                case FishingOutcome.Continue          ⇒ loop(newTimeout, msg :: seen)
-                case FishingOutcome.ContinueAndIgnore ⇒ loop(newTimeout, seen)
-                case _                                ⇒ ??? // cannot happen
-              }
-
+      val maybeMsg = Option(receiveOne(timeout))
+      maybeMsg match {
+        case Some(msg) ⇒
+          try {
+            fisher(msg) match {
+              case FishingOutcome.Complete    ⇒ (msg :: seen).reverse
+              case FishingOutcome.Fail(error) ⇒ throw new AssertionError(s"$error, hint: $hint")
+              case continue ⇒
+                val newTimeout =
+                  if (timeout.isFinite()) timeout - (System.nanoTime() - start).nanos
+                  else timeout
+                if (newTimeout.toMillis <= 0) {
+                  throw new AssertionError(s"timeout ($max) during fishForMessage, seen messages ${seen.reverse}, hint: $hint")
+                } else {
+
+                  continue match {
+                    case FishingOutcome.Continue          ⇒ loop(newTimeout, msg :: seen)
+                    case FishingOutcome.ContinueAndIgnore ⇒ loop(newTimeout, seen)
+                    case _                                ⇒ ??? // cannot happen
+                  }
+
+                }
             }
-        }
-      } catch {
-        case ex: MatchError ⇒ throw new AssertionError(
-          s"Unexpected message $msg while fishing for messages, " +
-            s"seen messages ${seen.reverse}, hint: $hint", ex)
+          } catch {
+            case ex: MatchError ⇒ throw new AssertionError(
+              s"Unexpected message $msg while fishing for messages, " +
+                s"seen messages ${seen.reverse}, hint: $hint", ex)
+          }
+
+        case None ⇒
+          throw new AssertionError(s"timeout ($max) during fishForMessage, seen messages ${seen.reverse}, hint: $hint")
       }
     }
 

diff --git a/...-actor-testkit-typed/src/test/scala/akka/actor/testkit/typed/scaladsl/TestProbeSpec.scala b/...-actor-testkit-typed/src/test/scala/akka/actor/testkit/typed/scaladsl/TestProbeSpec.scala
@@ -4,8 +4,8 @@
 
 package akka.actor.testkit.typed.scaladsl
 
-import scala.concurrent.duration._
 import akka.actor.typed.scaladsl.Behaviors
+import scala.concurrent.duration._
 import org.scalatest.WordSpecLike
 
 class TestProbeSpec extends ScalaTestWithActorTestKit with WordSpecLike {
@@ -84,6 +84,17 @@ class TestProbeSpec extends ScalaTestWithActorTestKit with WordSpecLike {
       }
     }
 
+    "throw an AssertionError when the fishing probe times out" in {
+      val probe = TestProbe[AnyRef]()
+
+      assertThrows[AssertionError] {
+        probe.fishForMessage(100.millis) { _ ⇒
+          Thread.sleep(150)
+          FishingOutcomes.complete
+        }
+      }
+    }
+
     "fail for unknown message when fishing for messages" in {
       val probe = TestProbe[String]()