awaitCond = awaitCond with better error reporting, see #3168

akka-docs/rst/java/code/docs/testkit/TestKitDocTest.java

@@ -187,6 +187,24 @@ public class TestKitDocTest {
     }};
     //#test-awaitCond
   }
+
+  @Test
+  public void demonstrateAwaitAssert() {
+    //#test-awaitAssert
+    new JavaTestKit(system) {{
+      getRef().tell(42, null);
+      new AwaitAssert(
+            duration("1 second"),  // maximum wait time
+            duration("100 millis") // interval at which to check the condition
+            ) {
+        // do not put code outside this method, will run afterwards
+        protected void check() {
+          assertEquals(msgAvailable(), true);
+        }
+      };
+    }};
+    //#test-awaitAssert
+  }
   
   @Test
   @SuppressWarnings("unchecked") // due to generic varargs

akka-docs/rst/java/testing.rst

@@ -282,6 +282,13 @@ code blocks:
     reception, the embedded condition can compute the boolean result from
     anything in scope.
 
+    * **AwaitAssert**
+
+    .. includecode:: code/docs/testkit/TestKitDocTest.java#test-awaitAssert
+
+    This general construct is not connected with the test kit’s message
+    reception, the embedded assert can check anything in scope.
+
 There are also cases where not all messages sent to the test kit are actually
 relevant to the test, but removing them would mean altering the actors under
 test. For this purpose it is possible to ignore certain messages:

akka-docs/rst/scala/testing.rst

@@ -304,6 +304,15 @@ with message flows:
     maximum defaults to the time remaining in the innermost enclosing
     :ref:`within <TestKit.within>` block.
 
+  * :meth:`awaitAssert(a: => Any, max: Duration, interval: Duration)`
+
+    Poll the given assert function every :obj:`interval` until it does not throw
+    an exception or the :obj:`max` duration is used up. If the timeout expires the 
+    last exception is thrown. The interval defaults to 100 ms and the maximum defaults
+    to the time remaining in the innermost enclosing :ref:`within <TestKit.within>` 
+    block.The interval defaults to 100 ms and the maximum defaults to the time 
+    remaining in the innermost enclosing :ref:`within <TestKit.within>` block.  
+
   * :meth:`ignoreMsg(pf: PartialFunction[AnyRef, Boolean])`
 
     :meth:`ignoreNoMsg`

akka-testkit/src/main/java/akka/testkit/JavaTestKit.java

@@ -14,7 +14,8 @@ import scala.concurrent.duration.Duration;
 import scala.concurrent.duration.FiniteDuration;
 
 /**
- * Java API for the TestProbe. Proper JavaDocs to come once JavaDoccing is implemented.
+ * Java API for the TestProbe. Proper JavaDocs to come once JavaDoccing is
+ * implemented.
  */
 public class JavaTestKit {
   private final TestProbe p;
@@ -30,13 +31,15 @@ public class JavaTestKit {
   public ActorSystem getSystem() {
     return p.system();
   }
-  
+
   static public FiniteDuration duration(String s) {
     final Duration ret = Duration.apply(s);
-    if (ret instanceof FiniteDuration) return (FiniteDuration) ret;
+    if (ret instanceof FiniteDuration)
-    else throw new IllegalArgumentException("duration() is only for finite durations, use Duration.Inf() and friends");
+      return (FiniteDuration) ret;
+    else
+      throw new IllegalArgumentException("duration() is only for finite durations, use Duration.Inf() and friends");
   }
-  
+
   public Duration dilated(Duration d) {
     return d.mul(TestKitExtension.get(p.system()).TestTimeFactor());
   }
@@ -137,7 +140,7 @@ public class JavaTestKit {
         }
       }, max, interval, p.awaitCond$default$4());
     }
-    
+
     public AwaitCond(Duration max, Duration interval, String message) {
       p.awaitCond(new AbstractFunction0<Object>() {
         public Object apply() {
@@ -147,6 +150,27 @@ public class JavaTestKit {
     }
   }
 
+  public abstract class AwaitAssert {
+    protected abstract void check();
+
+    public AwaitAssert() {
+      this(Duration.Undefined(), p.awaitAssert$default$3());
+    }
+
+    public AwaitAssert(Duration max) {
+      this(max, p.awaitAssert$default$3());
+    }
+
+    public AwaitAssert(Duration max, Duration interval) {
+      p.awaitAssert(new AbstractFunction0<Object>() {
+        public Object apply() {
+          check();
+          return null;
+        }
+      }, max, interval);
+    }
+  }
+
   public abstract class ExpectMsg<T> {
     private final T result;
 
@@ -159,8 +183,7 @@ public class JavaTestKit {
       try {
         result = match(received);
       } catch (JavaPartialFunction.NoMatchException ex) {
-        throw new AssertionError("while expecting '" + hint
+        throw new AssertionError("while expecting '" + hint + "' received unexpected: " + received);
-            + "' received unexpected: " + received);
       }
     }
 
@@ -200,13 +223,11 @@ public class JavaTestKit {
   }
 
   public Object[] expectMsgAllOf(Object... msgs) {
-    return (Object[]) p.expectMsgAllOf(Util.immutableSeq(msgs)).toArray(
+    return (Object[]) p.expectMsgAllOf(Util.immutableSeq(msgs)).toArray(Util.classTag(Object.class));
-        Util.classTag(Object.class));
   }
 
   public Object[] expectMsgAllOf(FiniteDuration max, Object... msgs) {
-    return (Object[]) p.expectMsgAllOf(max, Util.immutableSeq(msgs)).toArray(
+    return (Object[]) p.expectMsgAllOf(max, Util.immutableSeq(msgs)).toArray(Util.classTag(Object.class));
-        Util.classTag(Object.class));
   }
 
   @SuppressWarnings("unchecked")
@@ -254,12 +275,11 @@ public class JavaTestKit {
 
     @SuppressWarnings("unchecked")
     public ReceiveWhile(Class<T> clazz, Duration max, Duration idle, int messages) {
-      results = p.receiveWhile(max, idle, messages,
+      results = p.receiveWhile(max, idle, messages, new CachingPartialFunction<Object, T>() {
-          new CachingPartialFunction<Object, T>() {
+        public T match(Object msg) throws Exception {
-            public T match(Object msg) throws Exception {
+          return ReceiveWhile.this.match(msg);
-                return ReceiveWhile.this.match(msg);
+        }
-            }
+      }).toArray(Util.classTag(clazz));
-          }).toArray(Util.classTag(clazz));
     }
 
     protected RuntimeException noMatch() {
@@ -274,16 +294,16 @@ public class JavaTestKit {
 
   public abstract class EventFilter<T> {
     abstract protected T run();
-    
+
     private final Class<? extends Logging.LogEvent> clazz;
-    
+
     private String source = null;
     private String message = null;
     private boolean pattern = false;
     private boolean complete = false;
     private int occurrences = Integer.MAX_VALUE;
     private Class<? extends Throwable> exceptionType = null;
-    
+
     @SuppressWarnings("unchecked")
     public EventFilter(Class<?> clazz) {
       if (Throwable.class.isAssignableFrom(clazz)) {
@@ -291,13 +311,15 @@ public class JavaTestKit {
         exceptionType = (Class<? extends Throwable>) clazz;
       } else if (Logging.LogEvent.class.isAssignableFrom(clazz)) {
         this.clazz = (Class<? extends LogEvent>) clazz;
-      } else throw new IllegalArgumentException("supplied class must either be LogEvent or Throwable");
+      } else
+        throw new IllegalArgumentException("supplied class must either be LogEvent or Throwable");
     }
 
     public T exec() {
       akka.testkit.EventFilter filter;
       if (clazz == Logging.Error.class) {
-        if (exceptionType == null) exceptionType = Logging.noCause().getClass();
+        if (exceptionType == null)
+          exceptionType = Logging.noCause().getClass();
         filter = new ErrorFilter(exceptionType, source, message, pattern, complete, occurrences);
       } else if (clazz == Logging.Warning.class) {
         filter = new WarningFilter(source, message, pattern, complete, occurrences);
@@ -305,39 +327,40 @@ public class JavaTestKit {
         filter = new InfoFilter(source, message, pattern, complete, occurrences);
       } else if (clazz == Logging.Debug.class) {
         filter = new DebugFilter(source, message, pattern, complete, occurrences);
-      } else throw new IllegalArgumentException("unknown LogLevel " + clazz);
+      } else
+        throw new IllegalArgumentException("unknown LogLevel " + clazz);
       return filter.intercept(new AbstractFunction0<T>() {
         public T apply() {
           return run();
         }
       }, p.system());
     }
-    
+
     public EventFilter<T> message(String msg) {
       message = msg;
       pattern = false;
       complete = true;
       return this;
     }
-    
+
     public EventFilter<T> startsWith(String msg) {
       message = msg;
       pattern = false;
       complete = false;
       return this;
     }
-    
+
     public EventFilter<T> matches(String regex) {
       message = regex;
       pattern = true;
       return this;
     }
-    
+
     public EventFilter<T> from(String source) {
       this.source = source;
       return this;
     }
-    
+
     public EventFilter<T> occurrences(int number) {
       occurrences = number;
       return this;

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

@@ -4,7 +4,6 @@
 package akka.testkit
 
 import language.postfixOps
-
 import scala.annotation.{ varargs, tailrec }
 import scala.collection.immutable
 import scala.concurrent.duration._
@@ -14,6 +13,7 @@ import java.util.concurrent.atomic.AtomicInteger
 import akka.actor._
 import akka.actor.Actor._
 import akka.util.{ Timeout, BoxedType }
+import scala.util.control.NonFatal
 
 object TestActor {
   type Ignore = Option[PartialFunction[Any, Boolean]]
@@ -227,6 +227,38 @@ trait TestKitBase {
     poll(_max min interval)
   }
 
+  /**
+   * Await until the given assert does not throw an exception or the timeout
+   * expires, whichever comes first. If the timeout expires the last exception
+   * is thrown.
+   *
+   * If no timeout is given, take it from the innermost enclosing `within`
+   * block.
+   *
+   * Note that the timeout is scaled using Duration.dilated,
+   * which uses the configuration entry "akka.test.timefactor".
+   */
+  def awaitAssert(a: ⇒ Any, max: Duration = Duration.Undefined, interval: Duration = 100.millis) {
+    val _max = remainingOrDilated(max)
+    val stop = now + _max
+
+    @tailrec
+    def poll(t: Duration) {
+      val failed =
+        try { a; false } catch {
+          case NonFatal(e) ⇒
+            if (now >= stop) throw e
+            true
+        }
+      if (failed) {
+        Thread.sleep(t.toMillis)
+        poll((stop - now) min interval)
+      }
+    }
+
+    poll(_max min interval)
+  }
+
   /**
    * Execute code block while bounding its execution time between `min` and
    * `max`. `within` blocks may be nested. All methods in this trait which

akka-testkit/src/test/scala/akka/testkit/TestTimeSpec.scala

@@ -2,8 +2,9 @@ package akka.testkit
 
 import org.scalatest.matchers.MustMatchers
 import org.scalatest.{ BeforeAndAfterEach, WordSpec }
-import scala.concurrent.duration.Duration
+import scala.concurrent.duration._
 import com.typesafe.config.Config
+import org.scalatest.exceptions.TestFailedException
 
 @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
 class TestTimeSpec extends AkkaSpec(Map("akka.test.timefactor" -> 2.0)) with BeforeAndAfterEach {
@@ -20,6 +21,15 @@ class TestTimeSpec extends AkkaSpec(Map("akka.test.timefactor" -> 2.0)) with Bef
       diff must be < (target + 300000000l)
     }
 
+    "awaitAssert must throw correctly" in {
+      awaitAssert("foo" must be("foo"))
+      within(300.millis, 2.seconds) {
+        intercept[TestFailedException] {
+          awaitAssert("foo" must be("bar"), 500.millis)
+        }
+      }
+    }
+
   }
 
 }