Revert "Simplify log-wait-strategy (#977)" (#984)

This reverts commit b837e90.

Author: cristianrgreco

packages/testcontainers/src/wait-strategies/log-wait-strategy.test.ts

@@ -54,31 +54,4 @@ describe("LogWaitStrategy", { timeout: 180_000 }, () => {
 
     expect(await getRunningContainerNames()).not.toContain(containerName);
   });
-
-  it("should throw an error if the message is never received", async () => {
-    const containerName = `container-${new RandomUuid().nextUuid()}`;
-
-    await expect(
-      new GenericContainer("cristianrgreco/testcontainer:1.1.14")
-        .withName(containerName)
-        .withCommand("/bin/sh", "-c", 'echo "Ready"')
-        .withWaitStrategy(Wait.forLogMessage("unexpected"))
-        .start()
-    ).rejects.toThrowError(`Log stream ended and message "unexpected" was not received`);
-
-    expect(await getRunningContainerNames()).not.toContain(containerName);
-  });
-
-  it("does not matter if container does not send all content in a single line", async () => {
-    const container = await new GenericContainer("cristianrgreco/testcontainer:1.1.14")
-      .withCommand([
-        "node",
-        "-e",
-        "process.stdout.write('Hello '); setTimeout(() => process.stdout.write('World\\n'), 2000)",
-      ])
-      .withWaitStrategy(Wait.forLogMessage("Hello World"))
-      .start();
-
-    await container.stop();
-  });
 });

packages/testcontainers/src/wait-strategies/log-wait-strategy.ts

@@ -1,6 +1,5 @@
 import byline from "byline";
 import Dockerode from "dockerode";
-import { setTimeout } from "timers/promises";
 import { log } from "../common";
 import { getContainerRuntimeClient } from "../container-runtime";
 import { BoundPorts } from "../utils/bound-ports";
@@ -17,37 +16,46 @@ export class LogWaitStrategy extends AbstractWaitStrategy {
   }
 
   public async waitUntilReady(container: Dockerode.Container, boundPorts: BoundPorts, startTime?: Date): Promise<void> {
-    await Promise.race([this.handleTimeout(container.id), this.handleLogs(container, startTime)]);
-  }
-
-  async handleTimeout(containerId: string): Promise<void> {
-    await setTimeout(this.startupTimeout);
-    this.throwError(containerId, `Log message "${this.message}" not received after ${this.startupTimeout}ms`);
-  }
-
-  async handleLogs(container: Dockerode.Container, startTime?: Date): Promise<void> {
     log.debug(`Waiting for log message "${this.message}"...`, { containerId: container.id });
     const client = await getContainerRuntimeClient();
     const stream = await client.container.logs(container, { since: startTime ? startTime.getTime() / 1000 : 0 });
-
-    let matches = 0;
-    for await (const line of byline(stream)) {
-      if (this.matches(line)) {
-        if (++matches === this.times) {
-          return log.debug(`Log wait strategy complete`, { containerId: container.id });
+    return new Promise((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        const message = `Log message "${this.message}" not received after ${this.startupTimeout}ms`;
+        log.error(message, { containerId: container.id });
+        reject(new Error(message));
+      }, this.startupTimeout);
+
+      const comparisonFn: (line: string) => boolean = (line: string) => {
+        if (this.message instanceof RegExp) {
+          return this.message.test(line);
+        } else {
+          return line.includes(this.message);
         }
-      }
-    }
-
-    this.throwError(container.id, `Log stream ended and message "${this.message}" was not received`);
-  }
-
-  matches(line: string): boolean {
-    return this.message instanceof RegExp ? this.message.test(line) : line.includes(this.message);
-  }
-
-  throwError(containerId: string, message: string): void {
-    log.error(message, { containerId });
-    throw new Error(message);
+      };
+
+      let count = 0;
+      const lineProcessor = (line: string) => {
+        if (comparisonFn(line)) {
+          if (++count === this.times) {
+            stream.destroy();
+            clearTimeout(timeout);
+            log.debug(`Log wait strategy complete`, { containerId: container.id });
+            resolve();
+          }
+        }
+      };
+
+      byline(stream)
+        .on("data", lineProcessor)
+        .on("err", lineProcessor)
+        .on("end", () => {
+          stream.destroy();
+          clearTimeout(timeout);
+          const message = `Log stream ended and message "${this.message}" was not received`;
+          log.error(message, { containerId: container.id });
+          reject(new Error(message));
+        });
+    });
   }
 }