feat: introduce LoggingConsumer based on log crate (#682)

The first consumer provided and probably the most useful: allows you to
mirror container logs to test logs.

Later we can consider adding the same consumer but based on `Tracing`

testcontainers/src/core/logs/consumer.rs

@@ -2,6 +2,8 @@ use futures::{future::BoxFuture, FutureExt};
 
 use crate::core::logs::LogFrame;
 
+pub mod logging_consumer;
+
 /// Log consumer is a trait that allows to consume log frames.
 /// Consumers will be called for each log frame that is produced by the container for the whole lifecycle of the container.
 pub trait LogConsumer: Send + Sync {

testcontainers/src/core/logs/consumer/logging_consumer.rs

@@ -0,0 +1,56 @@
+use futures::{future::BoxFuture, FutureExt};
+
+use crate::core::logs::{consumer::LogConsumer, LogFrame};
+
+/// A consumer that logs the output of container with the [`log`] crate.
+///
+/// By default, both standard out and standard error will both be emitted at INFO level.
+#[derive(Debug)]
+pub struct LoggingConsumer {
+    stdout_level: log::Level,
+    stderr_level: log::Level,
+}
+
+impl LoggingConsumer {
+    /// Creates a new instance of the logging consumer.
+    pub fn new() -> Self {
+        Self {
+            stdout_level: log::Level::Info,
+            stderr_level: log::Level::Info,
+        }
+    }
+
+    /// Sets the log level for standard out. By default, this is `INFO`.
+    pub fn with_stdout_level(mut self, level: log::Level) -> Self {
+        self.stdout_level = level;
+        self
+    }
+
+    /// Sets the log level for standard error. By default, this is `INFO`.
+    pub fn with_stderr_level(mut self, level: log::Level) -> Self {
+        self.stderr_level = level;
+        self
+    }
+}
+
+impl Default for LoggingConsumer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl LogConsumer for LoggingConsumer {
+    fn accept<'a>(&'a self, record: &'a LogFrame) -> BoxFuture<'a, ()> {
+        async move {
+            match record {
+                LogFrame::StdOut(bytes) => {
+                    log::log!(self.stdout_level, "{}", String::from_utf8_lossy(bytes));
+                }
+                LogFrame::StdErr(bytes) => {
+                    log::log!(self.stderr_level, "{}", String::from_utf8_lossy(bytes));
+                }
+            }
+        }
+        .boxed()
+    }
+}

testcontainers/tests/async_runner.rs

@@ -4,7 +4,9 @@ use bollard::Docker;
 use reqwest::StatusCode;
 use testcontainers::{
     core::{
-        logs::LogFrame, wait::HttpWaitStrategy, CmdWaitFor, ExecCommand, IntoContainerPort, WaitFor,
+        logs::{consumer::logging_consumer::LoggingConsumer, LogFrame},
+        wait::HttpWaitStrategy,
+        CmdWaitFor, ExecCommand, IntoContainerPort, WaitFor,
     },
     runners::AsyncRunner,
     GenericImage, *,
@@ -181,6 +183,7 @@ async fn async_run_with_log_consumer() -> anyhow::Result<()> {
                 let _ = tx.send(());
             }
         })
+        .with_log_consumer(LoggingConsumer::new().with_stderr_level(log::Level::Error))
         .start()
         .await?;
     rx.recv()?; // notification from consumer

testcontainers/tests/sync_runner.rs

@@ -2,7 +2,11 @@
 
 use reqwest::StatusCode;
 use testcontainers::{
-    core::{wait::HttpWaitStrategy, CmdWaitFor, ExecCommand, Host, IntoContainerPort, WaitFor},
+    core::{
+        logs::{consumer::logging_consumer::LoggingConsumer, LogFrame},
+        wait::HttpWaitStrategy,
+        CmdWaitFor, ExecCommand, Host, IntoContainerPort, WaitFor,
+    },
     runners::SyncRunner,
     *,
 };
@@ -196,3 +200,21 @@ fn sync_run_exec() -> anyhow::Result<()> {
     assert_eq!(stderr, "stderr 1\nstderr 2\n");
     Ok(())
 }
+
+#[test]
+fn sync_run_with_log_consumer() -> anyhow::Result<()> {
+    let _ = pretty_env_logger::try_init();
+
+    let (tx, rx) = std::sync::mpsc::sync_channel(1);
+    let _container = HelloWorld
+        .with_log_consumer(move |frame: &LogFrame| {
+            // notify when the expected message is found
+            if String::from_utf8_lossy(frame.bytes()) == "Hello from Docker!\n" {
+                let _ = tx.send(());
+            }
+        })
+        .with_log_consumer(LoggingConsumer::new().with_stderr_level(log::Level::Error))
+        .start()?;
+    rx.recv()?; // notification from consumer
+    Ok(())
+}