finish EventFilter work and apply in three places

- fix memory visibility issue with occurrences counter
- add non-throwing awaitCond and use that for better error reporting
- move occurrence book-keeping (side-effect) out of PF guard, was
  evaluated multiple times (of course!)
- apply in cases where EventFilter.custom was used (one legitimate use
  case remains)

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

@@ -65,7 +65,7 @@ object FSMActorSpec {
 
     whenUnhandled {
       case Ev(msg) ⇒ {
-        log.info("unhandled event " + msg + " in state " + stateName + " with data " + stateData)
+        log.warning("unhandled event " + msg + " in state " + stateName + " with data " + stateData)
         unhandledLatch.open
         stay
       }
@@ -138,10 +138,7 @@ class FSMActorSpec extends AkkaSpec(Configuration("akka.actor.debug.fsm" -> true
       transitionCallBackLatch.await
       lockedLatch.await
 
-      filterEvents(EventFilter.custom {
+      EventFilter.warning(start = "unhandled event", occurrences = 1) intercept {
-        case Logging.Info(_: Lock, _) ⇒ true
-        case _                        ⇒ false
-      }) {
         lock ! "not_handled"
         unhandledLatch.await
       }
@@ -200,10 +197,7 @@ class FSMActorSpec extends AkkaSpec(Configuration("akka.actor.debug.fsm" -> true
       new TestKit(AkkaApplication("fsm event", AkkaApplication.defaultConfig ++
         Configuration("akka.loglevel" -> "DEBUG",
           "akka.actor.debug.fsm" -> true))) {
-        app.mainbus.publish(TestEvent.Mute(EventFilter.custom {
+        EventFilter.debug() intercept {
-          case _: Logging.Debug ⇒ true
-          case _                ⇒ false
-        }))
           val fsm = TestActorRef(new Actor with LoggingFSM[Int, Null] {
             startWith(1, null)
             when(1) {
@@ -236,6 +230,7 @@ class FSMActorSpec extends AkkaSpec(Configuration("akka.actor.debug.fsm" -> true
           app.mainbus.unsubscribe(testActor)
         }
       }
+    }
 
     "fill rolling event log and hand it out" in {
       val fsmref = TestActorRef(new Actor with LoggingFSM[Int, Int] {

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

@@ -106,11 +106,8 @@ class FSMTimingSpec extends AkkaSpec with ImplicitSender {
     }
 
     "notify unhandled messages" in {
-      filterEvents(EventFilter.custom {
+      filterEvents(EventFilter.warning("unhandled event Tick in state TestUnhandled", source = fsm, occurrences = 1),
-        case Logging.Warning(`fsm`, "unhandled event Tick in state TestUnhandled") ⇒ true
+        EventFilter.warning("unhandled event Unhandled(test) in state TestUnhandled", source = fsm, occurrences = 1)) {
-        case Logging.Warning(`fsm`, "unhandled event Unhandled(test) in state TestUnhandled") ⇒ true
-        case _ ⇒ false
-      }) {
           fsm ! TestUnhandled
           within(1 second) {
             fsm ! Tick

akka-testkit/src/main/scala/akka/testkit/TestEventListener.scala

@@ -53,6 +53,7 @@ object TestEvent {
  */
 abstract class EventFilter(occurrences: Int) {
 
+  @volatile // JMM does not guarantee visibility for non-final fields
   private var todo = occurrences
 
   /**
@@ -69,7 +70,7 @@ abstract class EventFilter(occurrences: Int) {
   }
 
   def awaitDone(max: Duration): Boolean = {
-    if (todo != Int.MaxValue && todo > 0) TestKit.awaitCond(todo == 0, max)
+    if (todo != Int.MaxValue && todo > 0) TestKit.awaitCond(todo == 0, max, noThrow = true)
     todo == Int.MaxValue || todo == 0
   }
 
@@ -82,7 +83,10 @@ abstract class EventFilter(occurrences: Int) {
     try {
       val result = code
       if (!awaitDone(app.AkkaConfig.TestEventFilterLeeway))
+        if (todo > 0)
           throw new AssertionError("Timeout waiting for " + todo + " messages on " + this)
+        else
+          throw new AssertionError("Received " + (-todo) + " messages too many on " + this)
       result
     } finally app.mainbus publish TestEvent.UnMute(this)
   }
@@ -438,12 +442,12 @@ class TestEventListener extends Logging.DefaultLogger {
 
   var filters: List[EventFilter] = Nil
 
-  override def receive: Actor.Receive = ({
+  override def receive = {
     case InitializeLogger(bus) ⇒ Seq(classOf[Mute], classOf[UnMute]) foreach (bus.subscribe(context.self, _))
     case Mute(filters)         ⇒ filters foreach addFilter
     case UnMute(filters)       ⇒ filters foreach removeFilter
-    case event: LogEvent if filter(event) ⇒
+    case event: LogEvent       ⇒ if (!filter(event)) print(event)
-  }: Actor.Receive) orElse super.receive
+  }
 
   def filter(event: LogEvent): Boolean = filters exists (f ⇒ try { f(event) } catch { case e: Exception ⇒ false })
 

akka-testkit/src/main/scala/akka/testkit/TestKit.scala

@@ -559,19 +559,24 @@ object TestKit {
    *
    * Note that the timeout is scaled using Duration.timeFactor.
    */
-  def awaitCond(p: ⇒ Boolean, max: Duration, interval: Duration = 100.millis) {
+  def awaitCond(p: ⇒ Boolean, max: Duration, interval: Duration = 100.millis, noThrow: Boolean = false): Boolean = {
     val stop = now + max
 
     @tailrec
-    def poll(t: Duration) {
+    def poll(): Boolean = {
       if (!p) {
-        assert(now < stop, "timeout " + max + " expired")
+        val toSleep = stop - now
-        Thread.sleep(t.toMillis)
+        if (toSleep <= Duration.Zero) {
-        poll((stop - now) min interval)
+          if (noThrow) false
+          else throw new AssertionError("timeout " + max + " expired")
+        } else {
+          Thread.sleep((toSleep min interval).toMillis)
+          poll()
         }
+      } else true
     }
 
-    poll(max min interval)
+    poll()
   }
 
   /**