More accurate log processing.

CHANGELOG.md

@@ -6,6 +6,7 @@ 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))
+- Fixed log processing: multibyte unicode, linebreaks and ASCII color codes. Color codes can be turned on with `withRemoveAnsiCodes(false)` ([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))

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<SELF extends BaseConsumer<SELF>> implements Consumer<OutputFrame> {
+    @Getter
+    @Setter
+    private boolean removeColorCodes = true;
+
+    public SELF withRemoveAnsiCodes(boolean removeAnsiCodes) {
+        this.removeColorCodes = removeAnsiCodes;
+        return (SELF) this;
+    }
+}

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

@@ -2,28 +2,45 @@
 
 
 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_LINE = new byte[0];
+
+    private static final Pattern ANSI_COLOR_PATTERN = Pattern.compile("\u001B\\[[0-9;]+m");
+
+    private static final String LINE_BREAK_REGEX = "((\\r?\\n)|(\\r))";
+
+    static final String LINE_BREAK_AT_END_REGEX = LINE_BREAK_REGEX + "$";
 
     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 +62,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");
-                } else {
-                    consumer.accept(outputFrame);
+                    LOGGER.error("got frame with type {}, for which no handler is configured", frame.getStreamType());
+                } else if (outputFrame.getBytes() != null && outputFrame.getBytes().length > 0) {
+                    if (frame.getStreamType() == StreamType.RAW) {
+                        processRawFrame(outputFrame, consumer);
+                    } else {
+                        processOtherFrame(outputFrame, consumer);
+                    }
                 }
             }
         }
@@ -63,8 +84,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 +108,72 @@ public void close() throws IOException {
     public CountDownLatch getCompletionLatch() {
         return completionLatch;
     }
+
+    private synchronized void processRawFrame(OutputFrame outputFrame, Consumer<OutputFrame> consumer) {
+        String utf8String = outputFrame.getUtf8String();
+        byte[] bytes = outputFrame.getBytes();
+
+        // 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.
+        int lastCharacterType = Character.getType(utf8String.charAt(utf8String.length() - 1));
+        if (lastCharacterType == 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) {
+        String utf8String = outputFrame.getUtf8String();
+
+        utf8String = processAnsiColorCodes(utf8String, consumer);
+        consumer.accept(new OutputFrame(outputFrame.getType(), utf8String.getBytes()));
+    }
+
+    private void normalizeLogLines(String utf8String, Consumer<OutputFrame> consumer) {
+        // Reformat strings to normalize new lines.
+        List<String> lines = new ArrayList<>(Arrays.asList(utf8String.split(LINE_BREAK_REGEX)));
+        if (lines.isEmpty()) {
+            consumer.accept(new OutputFrame(OutputFrame.OutputType.STDOUT, EMPTY_LINE));
+            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,19 @@ 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();
+        utf8String = utf8String.replaceAll(FrameConsumerResultCallback.LINE_BREAK_AT_END_REGEX, "");
+        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[] NEW_LINE = "\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(NEW_LINE);
+                }
                 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);