More accurate log processing.

CHANGELOG.md

@@ -6,9 +6,11 @@ All notable changes to this project will be documented in this file.
 ### Fixed
 - Fixed missing `commons-codec` dependency ([\#642](https://github.com/testcontainers/testcontainers-java/issues/642))
 - Fixed `HostPortWaitStrategy` throws `NumberFormatException` when port is exposed but not mapped ([\#640](https://github.com/testcontainers/testcontainers-java/issues/640))
+- More accurate log processing. The line breaks can be not at end of line, the frame's boundary can split multibyte unicode symbols in the middle. All of this issues is fixed ([PR \#643](https://github.com/testcontainers/testcontainers-java/pull/643))
 
 ### Changed
 - Support multiple HTTP status codes for HttpWaitStrategy ([\#630](https://github.com/testcontainers/testcontainers-java/issues/630))
+- Standard log consumers is are subclasses of `BaseConsumer` class. Other consumers can still only implement a `Consumer` interface, but in `BaseConsumer` class added new `removeColorCodes` property and `withRemoveAnsiCodes(boolean)` method, that can control the log preprocessor to remove/not remove ANSI color codes from log frame before send it to the `Consumer`. By default, preprocessor remove the color codes ([PR \#643](https://github.com/testcontainers/testcontainers-java/pull/643)) 
 
 ## [1.7.0] - 2018-04-07
 

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

@@ -0,0 +1,17 @@
+package org.testcontainers.containers.output;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.function.Consumer;
+
+public abstract class BaseConsumer<S extends BaseConsumer<S>> implements Consumer<OutputFrame> {
+    @Getter
+    @Setter
+    private boolean removeColorCodes = true;
+
+    public S withRemoveAnsiCodes(boolean removeAnsiCodes) {
+        this.removeColorCodes = removeAnsiCodes;
+        return (S) this;
+    }
+}

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

@@ -2,28 +2,41 @@
 
 
 import com.github.dockerjava.api.model.Frame;
+import com.github.dockerjava.api.model.StreamType;
 import com.github.dockerjava.core.async.ResultCallbackTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.function.Consumer;
+import java.util.regex.Pattern;
 
 /**
  * This class can be used as a generic callback for docker-java commands that produce Frames.
  */
 public class FrameConsumerResultCallback extends ResultCallbackTemplate<FrameConsumerResultCallback, Frame> {
 
-    private final static Logger LOGGER = LoggerFactory.getLogger(FrameConsumerResultCallback.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(FrameConsumerResultCallback.class);
+
+    private static final byte[] EMPTY_BYTES = new byte[0];
+
+    private static final Pattern ANSI_COLOR_PATTERN = Pattern.compile("\u001B\\[[0-9;]+m");
 
     private Map<OutputFrame.OutputType, Consumer<OutputFrame>> consumers;
 
     private CountDownLatch completionLatch = new CountDownLatch(1);
 
+    private StringBuilder logString = new StringBuilder();
+
+    private OutputFrame brokenFrame;
+
     public FrameConsumerResultCallback() {
         consumers = new HashMap<>();
     }
@@ -45,9 +58,13 @@ public void onNext(Frame frame) {
             if (outputFrame != null) {
                 Consumer<OutputFrame> consumer = consumers.get(outputFrame.getType());
                 if (consumer == null) {
-                    LOGGER.error("got frame with type " + frame.getStreamType() + ", for which no handler is configured");
+                    LOGGER.error("got frame with type {}, for which no handler is configured", frame.getStreamType());
                 } else {
-                    consumer.accept(outputFrame);
+                    if (frame.getStreamType() == StreamType.RAW) {
+                        processRawFrame(outputFrame, consumer);
+                    } else {
+                        processOtherFrame(outputFrame, consumer);
+                    }
                 }
             }
         }
@@ -63,8 +80,17 @@ public void onError(Throwable throwable) {
 
     @Override
     public void close() throws IOException {
+        OutputFrame lastLine = null;
+
+        if (logString.length() > 0) {
+            lastLine = new OutputFrame(OutputFrame.OutputType.STDOUT, logString.toString().getBytes());
+        }
+
         // send an END frame to every consumer... but only once per consumer.
         for (Consumer<OutputFrame> consumer : new HashSet<>(consumers.values())) {
+            if (lastLine != null) {
+                consumer.accept(lastLine);
+            }
             consumer.accept(OutputFrame.END);
         }
         super.close();
@@ -78,4 +104,79 @@ public void close() throws IOException {
     public CountDownLatch getCompletionLatch() {
         return completionLatch;
     }
+
+    private synchronized void processRawFrame(OutputFrame outputFrame, Consumer<OutputFrame> consumer) {
+        if (outputFrame != null) {
+            String utf8String = outputFrame.getUtf8String();
+            byte[] bytes = outputFrame.getBytes();
+
+            if (utf8String != null && !utf8String.isEmpty()) {
+                // Merging the strings by bytes to solve the problem breaking non-latin unicode symbols.
+                if (brokenFrame != null) {
+                    bytes = merge(brokenFrame.getBytes(), bytes);
+                    utf8String = new String(bytes);
+                    brokenFrame = null;
+                }
+                // Logger chunks can break the string in middle of multibyte unicode character.
+                // Backup the bytes to reconstruct proper char sequence with bytes from next frame.
+                if (Character.getType(utf8String.charAt(utf8String.length() - 1)) == Character.OTHER_SYMBOL) {
+                    brokenFrame = new OutputFrame(outputFrame.getType(), bytes);
+                    return;
+                }
+
+                utf8String = processAnsiColorCodes(utf8String, consumer);
+                normalizeLogLines(utf8String, consumer);
+            }
+        }
+    }
+
+    private synchronized void processOtherFrame(OutputFrame outputFrame, Consumer<OutputFrame> consumer) {
+        if (outputFrame != null) {
+            String utf8String = outputFrame.getUtf8String();
+
+            if (utf8String != null && !utf8String.isEmpty()) {
+                utf8String = processAnsiColorCodes(utf8String, consumer);
+                consumer.accept(new OutputFrame(outputFrame.getType(), utf8String.getBytes()));
+            }
+        }
+    }
+
+    private void normalizeLogLines(String utf8String, Consumer<OutputFrame> consumer) {
+        // Reformat strings to normalize enters.
+        List<String> lines = new ArrayList<>(Arrays.asList(utf8String.split("((\\r?\\n)|(\\r))")));
+        if (lines.isEmpty()) {
+            consumer.accept(new OutputFrame(OutputFrame.OutputType.STDOUT, EMPTY_BYTES));
+            return;
+        }
+        if (utf8String.startsWith("\n") || utf8String.startsWith("\r")) {
+            lines.add(0, "");
+        }
+        if (utf8String.endsWith("\n") || utf8String.endsWith("\r")) {
+            lines.add("");
+        }
+        for (int i = 0; i < lines.size() - 1; i++) {
+            String line = lines.get(i);
+            if (i == 0 && logString.length() > 0) {
+                line = logString.toString() + line;
+                logString.setLength(0);
+            }
+            consumer.accept(new OutputFrame(OutputFrame.OutputType.STDOUT, line.getBytes()));
+        }
+        logString.append(lines.get(lines.size() - 1));
+    }
+
+    private String processAnsiColorCodes(String utf8String, Consumer<OutputFrame> consumer) {
+        if (!(consumer instanceof BaseConsumer) || ((BaseConsumer) consumer).isRemoveColorCodes()) {
+            return ANSI_COLOR_PATTERN.matcher(utf8String).replaceAll("");
+        }
+        return utf8String;
+    }
+
+
+    private byte[] merge(byte[] str1, byte[] str2) {
+        byte[] mergedString = new byte[str1.length + str2.length];
+        System.arraycopy(str1, 0, mergedString, 0, str1.length);
+        System.arraycopy(str2, 0, mergedString, str1.length, str2.length);
+        return mergedString;
+    }
 }

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

@@ -2,18 +2,13 @@
 
 import org.slf4j.Logger;
 
-import java.util.function.Consumer;
-import java.util.regex.Pattern;
-
 /**
  * A consumer for container output that logs output to an SLF4J logger.
  */
-public class Slf4jLogConsumer implements Consumer<OutputFrame> {
+public class Slf4jLogConsumer extends BaseConsumer<Slf4jLogConsumer> {
     private final Logger logger;
     private String prefix = "";
 
-    private static final Pattern ANSI_CODE_PATTERN = Pattern.compile("\\[\\d[ABCD]");
-
     public Slf4jLogConsumer(Logger logger) {
         this.logger = logger;
     }
@@ -25,28 +20,18 @@ public Slf4jLogConsumer withPrefix(String prefix) {
 
     @Override
     public void accept(OutputFrame outputFrame) {
-        if (outputFrame != null) {
-            String utf8String = outputFrame.getUtf8String();
-
-            if (utf8String != null) {
-                OutputFrame.OutputType outputType = outputFrame.getType();
-                String message = utf8String.trim();
-
-                if (ANSI_CODE_PATTERN.matcher(message).matches()) {
-                    return;
-                }
-
-                switch (outputType) {
-                    case END:
-                        break;
-                    case STDOUT:
-                    case STDERR:
-                        logger.info("{}{}: {}", prefix, outputType, message);
-                        break;
-                    default:
-                        throw new IllegalArgumentException("Unexpected outputType " + outputType);
-                }
-            }
+        OutputFrame.OutputType outputType = outputFrame.getType();
+
+        String utf8String = outputFrame.getUtf8String();
+        switch (outputType) {
+            case END:
+                break;
+            case STDOUT:
+            case STDERR:
+                logger.info("{}{}: {}", prefix, outputType, utf8String);
+                break;
+            default:
+                throw new IllegalArgumentException("Unexpected outputType " + outputType);
         }
     }
 }

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

@@ -5,21 +5,26 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.Charset;
-import java.util.function.Consumer;
 
 /**
  * Created by rnorth on 26/03/2016.
  */
-public class ToStringConsumer implements Consumer<OutputFrame> {
+public class ToStringConsumer extends BaseConsumer<ToStringConsumer> {
+    private static final byte[] ENTER_BYTES = "\n".getBytes();
 
+    private boolean firstLine = true;
     private ByteArrayOutputStream stringBuffer = new ByteArrayOutputStream();
 
     @Override
     public void accept(OutputFrame outputFrame) {
         try {
             if (outputFrame.getBytes() != null) {
+                if (!firstLine) {
+                    stringBuffer.write(ENTER_BYTES);
+                }
                 stringBuffer.write(outputFrame.getBytes());
                 stringBuffer.flush();
+                firstLine = false;
             }
         } catch (IOException e) {
             throw new RuntimeException(e);

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

@@ -6,14 +6,13 @@
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 /**
  * A consumer for container output that buffers lines in a {@link java.util.concurrent.BlockingDeque} and enables tests
  * to wait for a matching condition.
  */
-public class WaitingConsumer implements Consumer<OutputFrame> {
+public class WaitingConsumer extends BaseConsumer<WaitingConsumer> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(WaitingConsumer.class);