Output Handling

/**
 * Represents a frame of output from a container.
 * Contains the output bytes, type (stdout/stderr), and timestamp.
 */
public class OutputFrame {
    /**
     * Constructor for creating an OutputFrame from output type and bytes.
     *
     * @param type the output type (STDOUT, STDERR, or END)
     * @param bytes the output bytes
     */
    public OutputFrame(OutputType type, byte[] bytes);

    /**
     * Create an OutputFrame from a docker-java Frame.
     *
     * @param frame the docker-java Frame object
     * @return OutputFrame instance
     */
    public static OutputFrame forFrame(Frame frame);

    /**
     * Get the output as a UTF-8 string.
     *
     * @return output as UTF-8 string
     */
    public String getUtf8String();

    /**
     * Get the output as a UTF-8 string without the trailing line ending.
     *
     * @return output without line ending
     */
    public String getUtf8StringWithoutLineEnding();

    /**
     * Get the raw output bytes.
     *
     * @return raw byte array
     */
    public byte[] getBytes();

    /**
     * Get the type of output (stdout, stderr, or end marker).
     *
     * @return output type
     */
    public OutputType getType();

    /**
     * Static constant representing end of stream marker.
     * Use this to check for or signal stream completion.
     */
    public static final OutputFrame END;

    /**
     * Output type enumeration.
     */
    public enum OutputType {
        /** Standard output */
        STDOUT,
        /** Standard error */
        STDERR,
        /** End of stream marker */
        END;

        /**
         * Create an OutputType from a docker-java StreamType.
         *
         * @param streamType the docker-java StreamType
         * @return corresponding OutputType
         */
        public static OutputType forStreamType(StreamType streamType);
    }
}

import org.testcontainers.containers.output.OutputFrame;
import java.util.function.Consumer;

// Process output frames
Consumer<OutputFrame> frameProcessor = frame -> {
    switch (frame.getType()) {
        case STDOUT:
            System.out.println("[OUT] " + frame.getUtf8String());
            break;
        case STDERR:
            System.err.println("[ERR] " + frame.getUtf8String());
            break;
        case END:
            System.out.println("Stream ended");
            break;
    }
};

/**
 * Base class for output consumers providing common functionality.
 * By default, ANSI escape codes are removed from output (removeColorCodes defaults to true).
 * Note: Internal field is named removeColorCodes, but the method parameter is removeAnsiCodes.
 *
 * @param <SELF> Self-referential generic type for fluent API
 */
public abstract class BaseConsumer<SELF extends BaseConsumer<SELF>>
    implements Consumer<OutputFrame> {

    /**
     * Configure whether to remove ANSI escape codes from output.
     * Defaults to true (ANSI codes are removed by default).
     *
     * @param removeAnsiCodes true to remove ANSI codes, false to keep them
     * @return this consumer for method chaining
     */
    public SELF withRemoveAnsiCodes(boolean removeAnsiCodes);

    /**
     * Get whether ANSI escape codes are removed from output.
     * Lombok @Getter generates this method from the removeColorCodes field.
     *
     * @return true if ANSI codes are removed, false if kept
     */
    public boolean isRemoveColorCodes();

    /**
     * Set whether to remove ANSI escape codes from output.
     * Lombok @Setter generates this method from the removeColorCodes field.
     *
     * @param removeColorCodes true to remove ANSI codes, false to keep them
     */
    public void setRemoveColorCodes(boolean removeColorCodes);

    /**
     * Accept and process an output frame.
     *
     * @param frame the output frame to process
     */
    @Override
    public abstract void accept(OutputFrame frame);
}

/**
 * Consumer that sends container output to an SLF4J logger.
 * By default, all output (both STDOUT and STDERR) is logged at INFO level.
 * When separateOutputStreams is true, STDOUT is logged at INFO level and STDERR at ERROR level.
 *
 * Prefix formatting behavior:
 * - With prefix and separateOutputStreams: format is "[prefix] : {message}"
 * - With prefix and without separateOutputStreams: format is "[prefix] {outputType}: {message}"
 * - Without prefix and separateOutputStreams: just "{message}"
 * - Without prefix and without separateOutputStreams: "{outputType}: {message}"
 * Note: Prefixes are automatically wrapped in brackets with a trailing space.
 */
public class Slf4jLogConsumer extends BaseConsumer<Slf4jLogConsumer> {
    /**
     * Create a log consumer with the specified logger.
     *
     * @param logger the SLF4J logger to use
     */
    public Slf4jLogConsumer(Logger logger);

    /**
     * Create a log consumer with separate output stream handling.
     *
     * @param logger the SLF4J logger to use
     * @param separateOutputStreams true to log stdout to INFO and stderr to ERROR
     */
    public Slf4jLogConsumer(Logger logger, boolean separateOutputStreams);

    /**
     * Add a prefix to all log messages.
     * The prefix will be automatically wrapped in brackets with a trailing space: "[prefix] "
     *
     * @param prefix the prefix string (will be formatted as "[prefix] ")
     * @return this consumer for method chaining
     */
    public Slf4jLogConsumer withPrefix(String prefix);

    /**
     * Add an MDC (Mapped Diagnostic Context) entry.
     *
     * @param key MDC key
     * @param value MDC value
     * @return this consumer for method chaining
     */
    public Slf4jLogConsumer withMdc(String key, String value);

    /**
     * Add multiple MDC (Mapped Diagnostic Context) entries.
     *
     * @param mdc map of MDC entries
     * @return this consumer for method chaining
     */
    public Slf4jLogConsumer withMdc(Map<String, String> mdc);

    /**
     * Enable separate output streams (stdout to INFO, stderr to ERROR).
     *
     * @return this consumer for method chaining
     */
    public Slf4jLogConsumer withSeparateOutputStreams();

    @Override
    public Slf4jLogConsumer withRemoveAnsiCodes(boolean removeAnsiCodes);

    @Override
    public void accept(OutputFrame frame);
}

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.utility.DockerImageName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// Send container logs to application logger
Logger logger = LoggerFactory.getLogger("container.postgres");

GenericContainer<?> postgres = new GenericContainer<>(DockerImageName.parse("postgres:15"))
    .withExposedPorts(5432)
    .withLogConsumer(new Slf4jLogConsumer(logger));

postgres.start();
// Container output is logged via SLF4J

// With ANSI code filtering
Logger appLogger = LoggerFactory.getLogger("container.app");

GenericContainer<?> app = new GenericContainer<>(DockerImageName.parse("myapp:latest"))
    .withExposedPorts(8080)
    .withLogConsumer(new Slf4jLogConsumer(appLogger)
        .withRemoveAnsiCodes(true));  // Remove color codes

// Use logger name based on container
GenericContainer<?> redis = new GenericContainer<>(DockerImageName.parse("redis:7.0"))
    .withExposedPorts(6379)
    .withLogConsumer(new Slf4jLogConsumer(
        LoggerFactory.getLogger("testcontainers.redis")));

/**
 * Consumer that collects output into a string buffer.
 * Useful for capturing output for assertions in tests.
 */
public class ToStringConsumer extends BaseConsumer<ToStringConsumer> {
    /**
     * Get the collected output as a UTF-8 string.
     *
     * @return all collected output
     */
    public String toUtf8String();

    /**
     * Get the collected output with a specific charset.
     *
     * @param charset the charset to use for decoding
     * @return all collected output decoded with the specified charset
     */
    public String toString(Charset charset);

    @Override
    public ToStringConsumer withRemoveAnsiCodes(boolean removeAnsiCodes);

    @Override
    public void accept(OutputFrame frame);
}

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.ToStringConsumer;
import org.testcontainers.utility.DockerImageName;

// Collect all output for assertion
ToStringConsumer consumer = new ToStringConsumer();

GenericContainer<?> container = new GenericContainer<>(DockerImageName.parse("alpine:latest"))
    .withCommand("sh", "-c", "echo 'Hello World' && echo 'Test Output'")
    .withLogConsumer(consumer);

container.start();

// Wait for container to finish
Thread.sleep(1000);

// Assert on collected output
String output = consumer.toUtf8String();
assertTrue(output.contains("Hello World"));
assertTrue(output.contains("Test Output"));

// Collect output without ANSI codes
ToStringConsumer cleanConsumer = new ToStringConsumer()
    .withRemoveAnsiCodes(true);

GenericContainer<?> colorfulApp = new GenericContainer<>(DockerImageName.parse("myapp:latest"))
    .withLogConsumer(cleanConsumer);

colorfulApp.start();
String cleanOutput = cleanConsumer.toUtf8String();
// Output has ANSI escape codes removed

/**
 * Consumer that waits for specific output patterns or stream completion.
 * Useful for synchronizing test execution with container events.
 */
public class WaitingConsumer extends BaseConsumer<WaitingConsumer> {
    /**
     * Get access to the underlying frame buffer.
     *
     * @return the LinkedBlockingDeque containing all output frames
     */
    public LinkedBlockingDeque<OutputFrame> getFrames();

    /**
     * Wait indefinitely until a predicate matches an output frame.
     *
     * @param predicate condition to wait for
     * @throws TimeoutException if timeout is reached
     */
    public void waitUntil(Predicate<OutputFrame> predicate) throws TimeoutException;

    /**
     * Wait until a predicate matches an output frame, with timeout.
     *
     * @param predicate condition to wait for
     * @param limit timeout value
     * @param limitUnit timeout unit
     * @throws TimeoutException if timeout is reached
     */
    public void waitUntil(Predicate<OutputFrame> predicate, int limit, TimeUnit limitUnit)
        throws TimeoutException;

    /**
     * Wait until a predicate matches an output frame N times.
     *
     * @param predicate condition to wait for
     * @param limit timeout value
     * @param limitUnit timeout unit
     * @param times number of times the predicate must match
     * @throws TimeoutException if timeout is reached
     */
    public void waitUntil(Predicate<OutputFrame> predicate, long limit, TimeUnit limitUnit, int times)
        throws TimeoutException;

    /**
     * Wait until the output stream ends.
     * Waits indefinitely (Long.MAX_VALUE nanoseconds).
     */
    public void waitUntilEnd();

    /**
     * Wait until the output stream ends, with timeout.
     *
     * @param limit timeout value
     * @param limitUnit timeout unit
     * @throws TimeoutException if timeout is reached
     */
    public void waitUntilEnd(long limit, TimeUnit limitUnit) throws TimeoutException;

    @Override
    public WaitingConsumer withRemoveAnsiCodes(boolean removeAnsiCodes);

    @Override
    public void accept(OutputFrame frame);
}

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.containers.output.WaitingConsumer;
import org.testcontainers.utility.DockerImageName;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

// Wait for specific log message
WaitingConsumer waitingConsumer = new WaitingConsumer();

GenericContainer<?> app = new GenericContainer<>(DockerImageName.parse("myapp:latest"))
    .withExposedPorts(8080)
    .withLogConsumer(waitingConsumer);

app.start();

// Wait for application to be ready
waitingConsumer.waitUntil(
    frame -> frame.getUtf8String().contains("Application started"),
    30,
    TimeUnit.SECONDS
);

// Now safe to proceed with tests
String appUrl = String.format("http://%s:%d", app.getHost(), app.getMappedPort(8080));

// Wait for specific error message
WaitingConsumer errorWatcher = new WaitingConsumer();

GenericContainer<?> container = new GenericContainer<>(DockerImageName.parse("test:latest"))
    .withLogConsumer(errorWatcher);

container.start();

try {
    errorWatcher.waitUntil(
        frame -> frame.getType() == OutputFrame.OutputType.STDERR &&
                 frame.getUtf8String().contains("FATAL ERROR"),
        10,
        TimeUnit.SECONDS
    );
    fail("Expected fatal error did not occur");
} catch (TimeoutException e) {
    // No error occurred - test passes
}

// Wait for container to finish execution
WaitingConsumer finishWatcher = new WaitingConsumer();

GenericContainer<?> oneShot = new GenericContainer<>(DockerImageName.parse("alpine:latest"))
    .withCommand("sh", "-c", "sleep 5 && echo 'Done'")
    .withLogConsumer(finishWatcher);

oneShot.start();

// Wait for output stream to end
finishWatcher.waitUntilEnd(10, TimeUnit.SECONDS);

// Container has finished execution
String logs = oneShot.getLogs();

/**
 * Docker-java callback adapter for frame consumers.
 * Provides low-level access to container output streams.
 * WARNING: This is an advanced class. Most users should use Slf4jLogConsumer,
 * ToStringConsumer, or WaitingConsumer instead.
 */
public class FrameConsumerResultCallback
    extends ResultCallbackTemplate<FrameConsumerResultCallback, Frame> {

    /**
     * Add a consumer for a specific output type.
     *
     * @param outputType the output type (STDOUT, STDERR, END)
     * @param consumer the consumer to handle frames of this type
     */
    public void addConsumer(OutputFrame.OutputType outputType, Consumer<OutputFrame> consumer);

    /**
     * Get the completion latch for waiting until output stream ends.
     *
     * @return CountDownLatch that counts down when stream completes
     */
    public CountDownLatch getCompletionLatch();

    /**
     * Process the next frame from the docker-java stream.
     *
     * @param frame the next frame to process
     */
    @Override
    public void onNext(Frame frame);

    /**
     * Handle errors from the docker-java stream.
     *
     * @param throwable the error that occurred
     */
    @Override
    public void onError(Throwable throwable);

    /**
     * Close the callback and release resources.
     */
    @Override
    public void close();
}

Need to log to SLF4J/Logback/Log4j?
├─ Yes → Use Slf4jLogConsumer
└─ No
   └─ Need to capture output for assertions?
      ├─ Yes → Use ToStringConsumer
      └─ No
         └─ Need to wait for specific output?
            ├─ Yes → Use WaitingConsumer
            └─ No
               └─ Need custom per-frame processing?
                  ├─ Yes → Implement Consumer<OutputFrame>
                  └─ No → Need docker-java callback integration?
                     └─ Yes → Use FrameConsumerResultCallback (advanced)

// Pattern 1: Debug logging during development
container.withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("container")));

// Pattern 2: Verify output in tests
ToStringConsumer toStringConsumer = new ToStringConsumer();
container.withLogConsumer(toStringConsumer);
container.start();
assertThat(toStringConsumer.toUtf8String()).contains("Server started");

// Pattern 3: Synchronize with container startup
WaitingConsumer waitingConsumer = new WaitingConsumer();
container.withLogConsumer(waitingConsumer);
container.start();
waitingConsumer.waitUntil(frame ->
    frame.getUtf8String().contains("ready to accept connections"), 30, TimeUnit.SECONDS);

// Pattern 4: Multiple consumers for different purposes
container.withLogConsumer(new Slf4jLogConsumer(logger))  // Log everything
         .withLogConsumer(toStringConsumer);              // Also capture for assertions

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.utility.DockerImageName;
import org.slf4j.LoggerFactory;
import java.util.function.Consumer;

public class MultiConsumerExample {

    @Test
    public void testWithMultipleConsumers() throws Exception {
        // Create multiple consumers
        Slf4jLogConsumer slf4jConsumer = new Slf4jLogConsumer(
            LoggerFactory.getLogger("test.container"));

        ToStringConsumer stringConsumer = new ToStringConsumer();

        WaitingConsumer waitingConsumer = new WaitingConsumer();

        // Combine consumers
        Consumer<OutputFrame> multiConsumer = frame -> {
            slf4jConsumer.accept(frame);      // Log to SLF4J
            stringConsumer.accept(frame);      // Collect to string
            waitingConsumer.accept(frame);     // Enable waiting
        };

        GenericContainer<?> app = new GenericContainer<>(
                DockerImageName.parse("myapp:latest"))
            .withExposedPorts(8080)
            .withLogConsumer(multiConsumer);

        app.start();

        // Wait for ready message
        waitingConsumer.waitUntil(
            frame -> frame.getUtf8String().contains("Server started"),
            30,
            TimeUnit.SECONDS
        );

        // Run tests
        // ...

        // Analyze collected output
        String allOutput = stringConsumer.toUtf8String();
        assertTrue(allOutput.contains("Server started"));
        assertFalse(allOutput.contains("ERROR"));
    }
}

import org.testcontainers.containers.output.BaseConsumer;
import org.testcontainers.containers.output.OutputFrame;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;

// Custom consumer that writes to a file
public class FileLogConsumer extends BaseConsumer<FileLogConsumer> {
    private final PrintWriter writer;
    private final DateTimeFormatter formatter;

    public FileLogConsumer(String filename) throws IOException {
        this.writer = new PrintWriter(new FileWriter(filename, true));
        this.formatter = DateTimeFormatter.ISO_INSTANT;
    }

    @Override
    public void accept(OutputFrame frame) {
        if (frame.getType() != OutputFrame.OutputType.END) {
            String timestamp = formatter.format(Instant.now());
            String type = frame.getType().name();
            String message = frame.getUtf8StringWithoutLineEnding();

            writer.println(String.format("[%s] %s: %s", timestamp, type, message));
            writer.flush();
        }
    }

    public void close() {
        writer.close();
    }
}

// Custom consumer that collects errors
public class ErrorCollector extends BaseConsumer<ErrorCollector> {
    private final List<String> errors = new ArrayList<>();

    @Override
    public void accept(OutputFrame frame) {
        if (frame.getType() == OutputFrame.OutputType.STDERR) {
            String message = frame.getUtf8StringWithoutLineEnding();
            if (message.contains("ERROR") || message.contains("Exception")) {
                errors.add(message);
            }
        }
    }

    public List<String> getErrors() {
        return new ArrayList<>(errors);
    }

    public boolean hasErrors() {
        return !errors.isEmpty();
    }
}

// Usage
ErrorCollector errorCollector = new ErrorCollector();
FileLogConsumer fileLogger = new FileLogConsumer("container.log");

Consumer<OutputFrame> combined = frame -> {
    errorCollector.accept(frame);
    fileLogger.accept(frame);
};

GenericContainer<?> container = new GenericContainer<>(DockerImageName.parse("myapp:latest"))
    .withLogConsumer(combined);

container.start();

// After test
assertFalse(errorCollector.hasErrors(),
    "Container had errors: " + errorCollector.getErrors());
fileLogger.close();

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.containers.output.WaitingConsumer;
import org.testcontainers.utility.DockerImageName;
import java.time.Duration;
import java.util.regex.Pattern;

public class PatternBasedOutputTest {

    @Test
    public void testDatabaseMigration() throws Exception {
        WaitingConsumer consumer = new WaitingConsumer();

        GenericContainer<?> app = new GenericContainer<>(
                DockerImageName.parse("myapp:latest"))
            .withExposedPorts(8080)
            .withEnv("RUN_MIGRATIONS", "true")
            .withLogConsumer(consumer);

        app.start();

        // Wait for migration start
        consumer.waitUntil(
            frame -> frame.getUtf8String().contains("Starting database migrations"),
            10,
            TimeUnit.SECONDS
        );

        // Wait for migration completion with pattern matching
        Pattern migrationPattern = Pattern.compile(".*Completed (\\d+) migrations.*");
        consumer.waitUntil(
            frame -> migrationPattern.matcher(frame.getUtf8String()).find(),
            2,
            TimeUnit.MINUTES
        );

        // Application is ready for testing
    }

    @Test
    public void testWorkerScaling() throws Exception {
        WaitingConsumer consumer = new WaitingConsumer();

        GenericContainer<?> worker = new GenericContainer<>(
                DockerImageName.parse("worker:latest"))
            .withEnv("WORKERS", "3")
            .withLogConsumer(consumer);

        worker.start();

        // Count how many workers started
        int workersStarted = 0;
        Pattern workerPattern = Pattern.compile(".*Worker (\\d+) ready.*");

        for (int i = 0; i < 3; i++) {
            consumer.waitUntil(
                frame -> workerPattern.matcher(frame.getUtf8String()).find(),
                10,
                TimeUnit.SECONDS
            );
            workersStarted++;
        }

        assertEquals(3, workersStarted, "All workers should have started");
    }
}

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.containers.output.ToStringConsumer;
import org.testcontainers.utility.DockerImageName;

public class DebuggingOutputTest {

    @Test
    public void testWithDetailedLogging() {
        ToStringConsumer consumer = new ToStringConsumer();

        GenericContainer<?> app = new GenericContainer<>(
                DockerImageName.parse("myapp:latest"))
            .withExposedPorts(8080)
            .withEnv("DEBUG", "true")
            .withEnv("LOG_LEVEL", "DEBUG")
            .withLogConsumer(consumer);

        try {
            app.start();

            // Run test that might fail
            // ...

        } catch (Exception e) {
            // Dump container output for debugging
            System.err.println("=== Container Output ===");
            System.err.println(consumer.toUtf8String());
            System.err.println("=== End Container Output ===");
            throw e;
        }
    }

    @Test
    public void testWithFilteredOutput() {
        // Only capture ERROR and WARN messages
        ToStringConsumer errorConsumer = new ToStringConsumer();

        Consumer<OutputFrame> filteringConsumer = frame -> {
            String message = frame.getUtf8String();
            if (message.contains("ERROR") || message.contains("WARN")) {
                errorConsumer.accept(frame);
            }
        };

        GenericContainer<?> container = new GenericContainer<>(
                DockerImageName.parse("myapp:latest"))
            .withLogConsumer(filteringConsumer);

        container.start();

        // Only errors and warnings are collected
        String criticalMessages = errorConsumer.toUtf8String();

        if (!criticalMessages.isEmpty()) {
            System.err.println("Critical messages detected:");
            System.err.println(criticalMessages);
        }
    }
}

import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.utility.DockerImageName;

public class RealTimeMonitoringTest {

    @Test
    public void testWithRealTimeMonitoring() throws InterruptedException {
        AtomicInteger lineCount = new AtomicInteger(0);
        AtomicBoolean errorDetected = new AtomicBoolean(false);

        Consumer<OutputFrame> monitor = frame -> {
            lineCount.incrementAndGet();

            String message = frame.getUtf8String();
            System.out.println("[CONTAINER] " + message);

            // Detect errors in real-time
            if (message.contains("ERROR") || message.contains("Exception")) {
                errorDetected.set(true);
                System.err.println("!!! ERROR DETECTED: " + message);
            }

            // Trigger actions based on output
            if (message.contains("Memory usage: 90%")) {
                System.err.println("!!! HIGH MEMORY WARNING");
            }
        };

        GenericContainer<?> app = new GenericContainer<>(
                DockerImageName.parse("myapp:latest"))
            .withExposedPorts(8080)
            .withLogConsumer(monitor);

        app.start();

        // Give container time to produce output
        Thread.sleep(5000);

        // Check monitoring results
        assertTrue(lineCount.get() > 0, "Container should have produced output");
        assertFalse(errorDetected.get(), "No errors should be detected");
    }
}