Fix StreamType.RAW frame processing (#1101)

* Issue #1100. Fixes processing logs of the RAW type.

* Issue #1100. Remove the redundant import.

* Issue #1100. Fixed formatting. Using the OneShotStartupCheckStrategy for faster tests.

* Issue #1100. Fixed generic use of GenericContainer.

* Issue #1100. Remove the redundant import.

* Issue #1100. Added timeouts for tests.

Author: vektory79

core/src/main/java/org/testcontainers/dockerclient/transport/okhttp/OkHttpInvocationBuilder.java

@@ -27,10 +27,7 @@
 import okhttp3.RequestBody;
 import okhttp3.Response;
 import okhttp3.internal.connection.RealConnection;
-import okio.BufferedSink;
-import okio.BufferedSource;
-import okio.Okio;
-import okio.Source;
+import okio.*;
 import org.jetbrains.annotations.Nullable;
 import org.testcontainers.DockerClientFactory;
 
@@ -354,12 +351,16 @@ public void accept(BufferedSource source) {
                     source.skip(3);
                     int payloadSize = source.readInt();
 
-                    if(!source.request(payloadSize)) {
-                        return;
-                    }
-                    byte[] payload = source.readByteArray(payloadSize);
+                    if (streamType != StreamType.RAW) {
+                        if (!source.request(payloadSize)) {
+                            return;
+                        }
+                        byte[] payload = source.readByteArray(payloadSize);
 
-                    resultCallback.onNext(new Frame(streamType, payload));
+                        resultCallback.onNext(new Frame(streamType, payload));
+                    } else {
+                        resultCallback.onNext(new Frame(streamType, source.readByteArray()));
+                    }
                 }
             } catch (Exception e) {
                 resultCallback.onError(e);

core/src/test/java/org/testcontainers/junit/OutputStreamWithTTYTest.java

@@ -0,0 +1,89 @@
+package org.testcontainers.junit;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.OutputFrame;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.output.ToStringConsumer;
+import org.testcontainers.containers.output.WaitingConsumer;
+import org.testcontainers.containers.startupcheck.OneShotStartupCheckStrategy;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+
+import static org.rnorth.visibleassertions.VisibleAssertions.*;
+import static org.testcontainers.containers.output.OutputFrame.OutputType.STDOUT;
+
+@Slf4j
+public class OutputStreamWithTTYTest {
+
+    @Rule
+    public GenericContainer container = new GenericContainer<>("alpine:3.2")
+        .withCommand("ls -1")
+        .withStartupCheckStrategy(new OneShotStartupCheckStrategy())
+        .withCreateContainerCmdModifier(command -> command.withTty(true));
+
+    @Rule
+    public Timeout globalTimeout = Timeout.seconds(10);
+
+    @Test
+    public void testFetchStdout() throws TimeoutException {
+        WaitingConsumer consumer = new WaitingConsumer();
+
+        container.followOutput(consumer, STDOUT);
+
+        consumer.waitUntil(frame -> frame.getType() == STDOUT && frame.getUtf8String().contains("home"), 4, TimeUnit.SECONDS);
+    }
+
+    @Test
+    public void testFetchStdoutWithTimeout() {
+        WaitingConsumer consumer = new WaitingConsumer();
+
+        container.followOutput(consumer, STDOUT);
+
+        assertThrows("a TimeoutException should be thrown", TimeoutException.class, () -> {
+            consumer.waitUntil(frame -> frame.getType() == STDOUT && frame.getUtf8String().contains("qqq"), 1, TimeUnit.SECONDS);
+            return true;
+        });
+    }
+
+    @Test
+    public void testFetchStdoutWithNoLimit() throws TimeoutException {
+        WaitingConsumer consumer = new WaitingConsumer();
+
+        container.followOutput(consumer, STDOUT);
+
+        consumer.waitUntil(frame -> frame.getType() == STDOUT && frame.getUtf8String().contains("home"));
+    }
+
+    @Test
+    public void testLogConsumer() throws TimeoutException {
+        WaitingConsumer waitingConsumer = new WaitingConsumer();
+        Slf4jLogConsumer logConsumer = new Slf4jLogConsumer(log);
+
+        Consumer<OutputFrame> composedConsumer = logConsumer.andThen(waitingConsumer);
+        container.followOutput(composedConsumer);
+
+        waitingConsumer.waitUntil(frame -> frame.getType() == STDOUT && frame.getUtf8String().contains("home"));
+    }
+
+    @Test
+    public void testToStringConsumer() throws TimeoutException {
+        WaitingConsumer waitingConsumer = new WaitingConsumer();
+        ToStringConsumer toStringConsumer = new ToStringConsumer();
+
+        Consumer<OutputFrame> composedConsumer = toStringConsumer.andThen(waitingConsumer);
+        container.followOutput(composedConsumer);
+
+        waitingConsumer.waitUntilEnd(4, TimeUnit.SECONDS);
+
+        String utf8String = toStringConsumer.toUtf8String();
+        assertTrue("the expected first value was found", utf8String.contains("home"));
+        assertTrue("the expected last value was found", utf8String.contains("media"));
+        assertFalse("a non-expected value was found", utf8String.contains("qqq"));
+    }
+}