Add LogMessageWaitStrategy (#322)

* Add LogMessageWaitStrategy

* Change LogMessageWaitStrategy to use matches()

* Change LogUtils to follow output since container start

* Add 'times' parameter to WaitingConsumer and LogMessageWaitStrategy

core/src/main/java/org/testcontainers/containers/output/WaitingConsumer.java

@@ -44,7 +44,7 @@ public LinkedBlockingDeque<OutputFrame> getFrames() {
      */
     public void waitUntil(Predicate<OutputFrame> predicate) throws TimeoutException {
         // ~2.9 million centuries ought to be enough for anyone
-        waitUntil(predicate, Long.MAX_VALUE);
+        waitUntil(predicate, Long.MAX_VALUE, 1);
     }
 
     /**
@@ -57,13 +57,30 @@ public void waitUntil(Predicate<OutputFrame> predicate) throws TimeoutException
      * @param limit     maximum time to wait
      * @param limitUnit maximum time to wait (units)
      */
-    public void waitUntil(Predicate<OutputFrame> predicate, long limit, TimeUnit limitUnit) throws TimeoutException {
+    public void waitUntil(Predicate<OutputFrame> predicate, int limit, TimeUnit limitUnit) throws TimeoutException {
+        waitUntil(predicate, limit, limitUnit, 1);
+    }
+
+    /**
+     * Wait until any frame (usually, line) of output matches the provided predicate.
+     * <p>
+     * Note that lines will often have a trailing newline character, and this is not stripped off before the
+     * predicate is tested.
+     *
+     * @param predicate a predicate to test against each frame
+     * @param limit     maximum time to wait
+     * @param limitUnit maximum time to wait (units)
+     * @param times     number of times the predicate has to match
+     */
+    public void waitUntil(Predicate<OutputFrame> predicate, long limit, TimeUnit limitUnit, int times) throws TimeoutException {
         long expiry = limitUnit.toMillis(limit) + System.currentTimeMillis();
 
-        waitUntil(predicate, expiry);
+        waitUntil(predicate, expiry, times);
     }
 
-    private void waitUntil(Predicate<OutputFrame> predicate, long expiry) throws TimeoutException {
+    private void waitUntil(Predicate<OutputFrame> predicate, long expiry, int times) throws TimeoutException {
+
+        int numberOfMatches = 0;
         while (System.currentTimeMillis() < expiry) {
             try {
                 OutputFrame frame = frames.pollLast(100, TimeUnit.MILLISECONDS);
@@ -72,7 +89,11 @@ private void waitUntil(Predicate<OutputFrame> predicate, long expiry) throws Tim
                     LOGGER.debug("{}: {}", frame.getType(), frame.getUtf8String());
 
                     if (predicate.test(frame)) {
-                        return;
+                        numberOfMatches++;
+
+                        if (numberOfMatches == times) {
+                            return;
+                        }
                     }
                 }
 

core/src/main/java/org/testcontainers/containers/wait/LogMessageWaitStrategy.java

@@ -0,0 +1,44 @@
+package org.testcontainers.containers.wait;
+
+import org.testcontainers.containers.ContainerLaunchException;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.OutputFrame;
+import org.testcontainers.containers.output.WaitingConsumer;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+
+/**
+ * Waits until containers logs expected content.
+ */
+public class LogMessageWaitStrategy extends GenericContainer.AbstractWaitStrategy {
+    private String regEx;
+
+    private int times = 1;
+
+    @Override
+    protected void waitUntilReady() {
+        WaitingConsumer waitingConsumer = new WaitingConsumer();
+        container.followOutput(waitingConsumer);
+
+        Predicate<OutputFrame> waitPredicate = outputFrame ->
+                outputFrame.getUtf8String().matches(regEx);
+
+        try {
+            waitingConsumer.waitUntil(waitPredicate, startupTimeout.getSeconds(), TimeUnit.SECONDS, times);
+        } catch (TimeoutException e) {
+            throw new ContainerLaunchException("Timed out waiting for log output matching '" + regEx + "'");
+        }
+    }
+
+    public LogMessageWaitStrategy withRegEx(String regEx) {
+        this.regEx = regEx;
+        return this;
+    }
+
+    public LogMessageWaitStrategy withTimes(int times) {
+        this.times = times;
+        return this;
+    }
+}

core/src/main/java/org/testcontainers/utility/LogUtils.java

@@ -23,7 +23,8 @@ public void followOutput(DockerClient dockerClient, String containerId,
                              Consumer<OutputFrame> consumer, OutputFrame.OutputType... types) {
 
         final LogContainerCmd cmd = dockerClient.logContainerCmd(containerId)
-                .withFollowStream(true);
+                .withFollowStream(true)
+                .withSince(0);
 
         final FrameConsumerResultCallback callback = new FrameConsumerResultCallback();
         for (OutputFrame.OutputType type : types) {

core/src/test/java/org/testcontainers/junit/wait/LogMessageWaitStrategyTest.java

@@ -0,0 +1,43 @@
+package org.testcontainers.junit.wait;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Test;
+import org.testcontainers.containers.wait.LogMessageWaitStrategy;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Tests for {@link LogMessageWaitStrategy}.
+ */
+public class LogMessageWaitStrategyTest extends AbstractWaitStrategyTest<LogMessageWaitStrategy> {
+
+    private static final String READY_MESSAGE = "I'm ready!";
+
+    @Test
+    public void testWaitUntilReady_Success() {
+        waitUntilReadyAndSucceed("echo -e \"" + READY_MESSAGE + "\";" +
+                "echo -e \"foobar\";" +
+                "echo -e \"" + READY_MESSAGE + "\";" +
+                "sleep 300");
+    }
+
+    @Test
+    public void testWaitUntilReady_Timeout() {
+        waitUntilReadyAndTimeout("echo -e \"" + READY_MESSAGE + "\";" +
+                "echo -e \"foobar\";" +
+                "sleep 300");
+    }
+
+    @NotNull
+    @Override
+    protected LogMessageWaitStrategy buildWaitStrategy(AtomicBoolean ready) {
+
+        return new LogMessageWaitStrategy() {
+            @Override
+            protected void waitUntilReady() {
+                super.waitUntilReady();
+                ready.set(true);
+            }
+        }.withRegEx(".*ready.*\\s").withTimes(2);
+    }
+}