Merge branch 'main' into socat-container

Author: cristianrgreco

packages/modules/kafka/src/kafka-container.test.ts

@@ -148,7 +148,7 @@ describe("KafkaContainer", { timeout: 240_000 }, () => {
 
       const kafkaCliContainer = await new GenericContainer(KAFKA_IMAGE)
         .withNetwork(network)
-        .withCommand(["bash", "-c", "echo 'START'; sleep infinity"])
+        .withCommand(["bash", "-c", "sleep infinity"])
         .withCopyFilesToContainer([
           {
             source: path.resolve(certificatesDir, "kafka.client.truststore.pem"),

packages/modules/localstack/src/localstack-container.test.ts

@@ -51,7 +51,7 @@ describe("LocalStackContainer", { timeout: 180_000 }, () => {
     const awsCliInDockerNetwork = await new GenericContainer("amazon/aws-cli:2.7.27")
       .withNetwork(network)
       .withEntrypoint(["bash"])
-      .withCommand(["-c", "echo 'START'; sleep infinity"])
+      .withCommand(["-c", "sleep infinity"])
       .withEnvironment({
         AWS_ACCESS_KEY_ID: "test",
         AWS_SECRET_ACCESS_KEY: "test",

packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts

@@ -161,24 +161,35 @@ export class DockerContainerClient implements ContainerClient {
     }
   }
 
-  async logs(container: Container, opts?: ContainerLogsOptions): Promise<Readable> {
-    try {
-      log.debug(`Fetching container logs...`, { containerId: container.id });
-      const stream = (await container.logs({
+  logs(container: Container, opts?: ContainerLogsOptions): Promise<Readable> {
+    log.debug(`Fetching container logs...`, { containerId: container.id });
+
+    const proxyStream = new PassThrough();
+    proxyStream.setEncoding("utf8");
+
+    container
+      .logs({
         follow: true,
         stdout: true,
         stderr: true,
         tail: opts?.tail ?? -1,
         since: opts?.since ?? 0,
-      })) as IncomingMessage;
-      stream.socket.unref();
-      const demuxedStream = this.demuxStream(container.id, stream);
-      log.debug(`Fetched container logs`, { containerId: container.id });
-      return demuxedStream;
-    } catch (err) {
-      log.error(`Failed to fetch container logs: ${err}`, { containerId: container.id });
-      throw err;
-    }
+      })
+      .then(async (stream) => {
+        const actualLogStream = stream as IncomingMessage;
+        actualLogStream.socket?.unref();
+
+        const demuxedStream = await this.demuxStream(container.id, actualLogStream);
+        demuxedStream.pipe(proxyStream);
+        demuxedStream.on("error", (err) => proxyStream.emit("error", err));
+        demuxedStream.on("end", () => proxyStream.end());
+      })
+      .catch((err) => {
+        log.error(`Failed to fetch container logs: ${err}`, { containerId: container.id });
+        proxyStream.end();
+      });
+
+    return Promise.resolve(proxyStream);
   }
 
   async exec(container: Container, command: string[], opts?: Partial<ExecOptions>): Promise<ExecResult> {