fix: make sure log consumers are stopped before deleting container

testcontainers/src/core/containers/async_container.rs

@@ -1,5 +1,6 @@
 use std::{fmt, ops::Deref, sync::Arc};
 
+use tokio::task::JoinHandle;
 use tokio_stream::StreamExt;
 
 #[cfg(feature = "host-port-exposure")]
@@ -32,6 +33,7 @@ pub struct ContainerAsync<I: Image> {
     pub(super) raw: raw::RawContainer,
     image: ContainerRequest<I>,
     network: Option<Arc<Network>>,
+    log_handler: Option<JoinHandle<()>>,
     dropped: bool,
     #[cfg(feature = "host-port-exposure")]
     host_port_exposure: Option<HostPortExposure>,
@@ -80,10 +82,11 @@ where
         let reuse = container_req.reuse();
 
         let log_consumers = std::mem::take(&mut container_req.log_consumers);
-        let container = ContainerAsync {
+        let mut container = ContainerAsync {
             raw: raw::RawContainer::new(id, docker_client),
             image: container_req,
             network,
+            log_handler: None,
             dropped: false,
             #[cfg(feature = "host-port-exposure")]
             host_port_exposure,
@@ -94,7 +97,7 @@ where
         if !log_consumers.is_empty() {
             let mut logs = container.docker_client().logs(container.id(), true);
             let container_id = container.id().to_string();
-            tokio::spawn(async move {
+            let log_handler = tokio::spawn(async move {
                 while let Some(result) = logs.next().await {
                     match result {
                         Ok(record) => {
@@ -111,6 +114,8 @@ where
                     }
                 }
             });
+
+            container.log_handler.replace(log_handler);
         }
 
         container
@@ -247,13 +252,22 @@ where
             let id = self.id().to_string();
             let client = self.docker_client().clone();
             let command = self.docker_client().config.command();
+            let log_handler = self.log_handler.take();
 
             let drop_task = async move {
                 log::trace!("Drop was called for container {id}, cleaning up");
                 match command {
                     env::Command::Remove => {
+                        if let Some(handle) = log_handler {
+                            if let Err(e) = client.stop(&id, None).await {
+                                log::error!("Failed to stop container on drop: {e}")
+                            }
+                            if let Err(e) = handle.await {
+                                log::error!("Failed to wait log consumers to finish: {e}");
+                            }
+                        }
                         if let Err(e) = client.rm(&id).await {
-                            log::error!("Failed to remove container on drop: {}", e);
+                            log::error!("Failed to remove container on drop: {e}");
                         }
                     }
                     env::Command::Keep => {}