Fix TcpProxy for non HTTP communication

examples/rental-model/advanced/tcp-proxy/server.js

@@ -0,0 +1,39 @@
+/* eslint-disable */
+const http = require("http");
+
+(async function main() {
+  const PORT = parseInt(process.env["PORT"] ?? "80");
+
+  // Increase the value if you want to test long response/liveliness scenarios
+  const SIMULATE_DELAY_SEC = parseInt(process.env["SIMULATE_DELAY_SEC"] ?? "0");
+
+  const respond = (res) => {
+    res.writeHead(200);
+    res.end("Hello Golem!");
+  };
+
+  const app = http.createServer((req, res) => {
+    if (SIMULATE_DELAY_SEC > 0) {
+      setTimeout(() => {
+        respond(res);
+      }, SIMULATE_DELAY_SEC * 1000);
+    } else {
+      respond(res);
+    }
+  });
+
+  const server = app.listen(PORT, () => console.log(`HTTP server started at "http://localhost:${PORT}"`));
+
+  const shutdown = () => {
+    server.close((err) => {
+      if (err) {
+        console.error("Server close encountered an issue", err);
+      } else {
+        console.log("Server closed successfully");
+      }
+    });
+  };
+
+  process.on("SIGINT", shutdown);
+  process.on("SIGTERM", shutdown);
+})();

examples/rental-model/advanced/tcp-proxy/tcp-proxy.ts

@@ -0,0 +1,86 @@
+import { GolemNetwork, waitFor } from "@golem-sdk/golem-js";
+import { pinoPrettyLogger } from "@golem-sdk/pino-logger";
+
+(async () => {
+  const logger = pinoPrettyLogger({
+    level: "info",
+  });
+  const glm = new GolemNetwork({
+    logger,
+  });
+
+  try {
+    await glm.connect();
+
+    const network = await glm.createNetwork({
+      ip: "10.0.0.0/24",
+    });
+
+    const rental = await glm.oneOf({
+      order: {
+        demand: {
+          workload: {
+            imageTag: "golem/node:20-alpine",
+            capabilities: ["vpn"],
+          },
+        },
+        market: {
+          rentHours: 0.25,
+          pricing: {
+            model: "burn-rate",
+            avgGlmPerHour: 1,
+          },
+        },
+        network,
+      },
+    });
+
+    const PORT_ON_PROVIDER = 80;
+    const PORT_ON_REQUESTOR = 8080;
+
+    const exe = await rental.getExeUnit();
+
+    // Install the server script
+    await exe.uploadFile(`./rental-model/advanced/tcp-proxy/server.js`, "/golem/work/server.js");
+
+    // Start the server process on the provider
+    const server = await exe.runAndStream(`PORT=${PORT_ON_PROVIDER} node /golem/work/server.js`);
+
+    server.stdout.subscribe((data) => console.log("provider>", data));
+    server.stderr.subscribe((data) => console.error("provider>", data));
+
+    // Create a proxy instance
+    const proxy = exe.createTcpProxy(PORT_ON_PROVIDER);
+    proxy.events.on("error", (error) => console.error("TcpProxy reported an error:", error));
+
+    // Start listening and expose the port on your requestor machine
+    await proxy.listen(PORT_ON_REQUESTOR);
+    console.log(`Server Proxy listen at http://localhost:${PORT_ON_REQUESTOR}`);
+
+    let isClosing = false;
+    const stopServer = async () => {
+      if (isClosing) {
+        console.log("Already closing, ignoring subsequent shutdown request");
+        return;
+      }
+
+      isClosing = true;
+
+      console.log("Shutting down gracefully");
+      await proxy.close();
+    };
+
+    process.on("SIGINT", () => {
+      stopServer()
+        .then(() => rental.stopAndFinalize())
+        .then(() => logger.info("Shutdown routine completed"))
+        .catch((err) => logger.error("Failed to shutdown cleanly", err));
+    });
+
+    await waitFor(() => server.isFinished());
+  } catch (error) {
+    logger.error("Failed to run the example", error);
+  } finally {
+    await glm.disconnect();
+  }
+})().catch(console.error);

src/activity/exe-unit/exe-unit.ts

@@ -20,7 +20,7 @@ import { RemoteProcess } from "./process";
 import { GolemWorkError, WorkErrorCode } from "./error";
 import { GolemAbortError, GolemConfigError, GolemTimeoutError } from "../../shared/error/golem-error";
 import { Agreement, ProviderInfo } from "../../market";
-import { TcpProxy } from "../../network/tcpProxy";
+import { TcpProxy } from "../../network/tcp-proxy";
 import { ExecutionOptions, ExeScriptExecutor } from "../exe-script-executor";
 import { lastValueFrom, tap, toArray } from "rxjs";
 

src/activity/exe-unit/index.ts

@@ -1,4 +1,4 @@
 export { ExeUnit, LifecycleFunction, ExeUnitOptions } from "./exe-unit";
 export { Batch } from "./batch";
 export { GolemWorkError, WorkErrorCode } from "./error";
-export { TcpProxy } from "../../network/tcpProxy";
+export { TcpProxy } from "../../network/tcp-proxy";

src/index.ts

@@ -10,7 +10,7 @@ export * from "./activity";
 
 // Necessary domain entities for users to consume
 export * from "./shared/error/golem-error";
-export * from "./network/tcpProxy";
+export * from "./network/tcp-proxy";
 
 // Internals
 export * from "./shared/utils";

src/market/market.module.test.ts

@@ -12,7 +12,7 @@ import { Allocation, IPaymentApi } from "../payment";
 import { INetworkApi, NetworkModule } from "../network";
 import { DraftOfferProposalPool } from "./draft-offer-proposal-pool";
 import { Agreement, AgreementEvent, ProviderInfo } from "./agreement";
-import { waitAndCall, waitForCondition } from "../shared/utils/wait";
+import { waitAndCall, waitFor } from "../shared/utils/wait";
 import { MarketOrderSpec } from "../golem-network";
 import { GolemAbortError } from "../shared/error/golem-error";
 
@@ -347,7 +347,7 @@ describe("Market module", () => {
         });
       });
 
-      await waitForCondition(() => draftListener.mock.calls.length > 0);
+      await waitFor(() => draftListener.mock.calls.length > 0);
       testSub.unsubscribe();
 
       expect(draftListener).toHaveBeenCalledWith(draftProposal);