diff --git a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts index 0a18e03c9..50a5c75c0 100644 --- a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts +++ b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts @@ -2,6 +2,29 @@ import { PassThrough, Readable } from "stream"; import { DockerContainerClient } from "./docker-container-client"; describe("DockerContainerClient", () => { + describe("logs", () => { + it("should destroy the Docker stream when the consumer closes the log stream", async () => { + const actualLogStream = new PassThrough(); + const demuxStream = vi.fn(); + const container = { + id: "container-id", + logs: vi.fn(async () => actualLogStream), + }; + const dockerode = { + modem: { demuxStream }, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + } as any; + const client = new DockerContainerClient(dockerode); + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const stream = await client.logs(container as any); + await vi.waitFor(() => expect(demuxStream).toHaveBeenCalledOnce()); + stream.destroy(); + + await vi.waitFor(() => expect(actualLogStream.destroyed).toBe(true)); + }); + }); + describe("exec", () => { it("should not truncate output when the demuxed streams flush after the raw stream ends", async () => { const payload = "the-final-line-that-must-not-be-truncated\n"; diff --git a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts index ada9def7a..ccbcd8ce4 100644 --- a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts +++ b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts @@ -194,6 +194,11 @@ export class DockerContainerClient implements ContainerClient { actualLogStream.socket?.unref(); const demuxedStream = await this.demuxStream(container.id, actualLogStream); + if (proxyStream.destroyed) { + demuxedStream.destroy(); + return; + } + proxyStream.once("close", () => demuxedStream.destroy()); demuxedStream.pipe(proxyStream); demuxedStream.on("error", (err) => proxyStream.emit("error", err)); demuxedStream.on("end", () => proxyStream.end());