v0.3.21-beta.1

Author: viktormarinho

channel.ts

@@ -179,14 +179,19 @@ export const makeReadableStream = (
 ): ReadableStream<Uint8Array> => {
   return new ReadableStream({
     async start(controller) {
-      for await (const content of ch.recv(signal)) {
-        controller.enqueue(content);
+      try {
+        for await (const content of ch.recv(signal)) {
+          controller.enqueue(content);
+        }
+        controller.close();
+      } catch (error) {
+        // Handle cancellation gracefully
+        if (signal?.aborted || error instanceof ClosedChannelError) {
+          controller.close();
+        } else {
+          controller.error(error);
+        }
       }
-      // Uncomment if necessary. this will send a signal to the controller
-      // if (signal?.aborted) {
-      //   controller.error(new Error("aborted"));
-      // }
-      controller.close();
     },
     cancel() {
       ch.close();
@@ -201,15 +206,23 @@ export const makeChanStream = (
   // Consume the transformed stream to trigger the pipeline
   const reader = stream.getReader();
   const processStream = async () => {
-    while (true) {
-      const { done, value } = await reader.read();
-      if (done) break;
-      await chan.send(value);
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        await chan.send(value);
+      }
+      chan.close();
+    } catch (err) {
+      // Handle errors gracefully, especially for Node.js/undici
+      if (!chan.signal.aborted) {
+        console.error("error processing stream", err);
+        chan.close();
+      }
     }
-    chan.close();
   };
   processStream().catch((err) => {
-    if (!err?.target?.aborted) {
+    if (!chan.signal.aborted) {
       console.error("error processing stream", err);
     }
   });

handlers.client.ts

@@ -51,6 +51,7 @@ const onRequestStart: ServerMessageHandler<RequestStartMessage> = async (
     await handleWebSocket(message, state);
     return;
   }
+  console.log(`[req-start] ${message.method} ${message.url} hasBody: ${message.hasBody}`);
   const abortCtrl = new AbortController();
   state.requests[message.id] = { abortCtrl };
   if (!message.hasBody) {
@@ -69,6 +70,11 @@ const onRequestStart: ServerMessageHandler<RequestStartMessage> = async (
       state.ch.out,
       abortCtrl.signal,
     ).catch(ignoreIfClosed).finally(() => {
+      // Ensure the body channel is closed when the request is cleaned up
+      const req = state.requests[message.id];
+      if (req?.body) {
+        req.body.close();
+      }
       delete state.requests[message.id];
     });
   }
@@ -88,6 +94,7 @@ const onRequestData: ServerMessageHandler<RequestDataMessage> = async (
     console.info("[req-data] req not found", message.id);
     return;
   }
+  console.log(`[req-data] ${message.id} chunk size: ${message.chunk.length}`);
   // @ts-ignore: bodyData is a Uint8Array
   await reqBody.send?.(message.chunk);
 };
@@ -113,10 +120,12 @@ const onRequestDataEnd: ServerMessageHandler<RequestDataEndMessage> = (
   state,
   message,
 ) => {
+  console.log(`[req-data-end] ${message.id}`);
   const reqBody = state.requests[message.id]?.body;
   if (!reqBody) {
     return;
   }
+  // Close the body channel to signal end of request data
   reqBody.close();
 };
 
@@ -239,20 +248,29 @@ async function doFetch(
   // Read from the stream
   const signal = link(clientCh.signal, reqSignal);
   try {
+    // For Node.js with undici, we need to handle the body differently
+    const fetchOptions: RequestInit = {
+      ...state.client ? { client: state.client } : {},
+      redirect: "manual",
+      method: request.method,
+      headers: state.client
+        ? { ...request.headers, host: request.domain }
+        : request.headers,
+      signal,
+    };
+
+    // Only add body and duplex if we actually have a body
+    if (request.body) {
+      fetchOptions.body = request.body;
+      // Remove duplex option for Node.js/undici compatibility
+      // fetchOptions.duplex = "half";
+    }
+
     const response = await fetch(
       new URL(request.url, state.localAddr),
-      {
-        ...state.client ? { client: state.client } : {},
-        redirect: "manual",
-        method: request.method,
-        headers: state.client
-          ? { ...request.headers, host: request.domain }
-          : request.headers,
-        body: request.body,
-        ...(request.body ? { duplex: "half" } : {}),
-        signal,
-      },
+      fetchOptions,
     );
+    console.log("response", response);
 
     const headers: Record<string, Array<string>> = {};
 
@@ -289,6 +307,7 @@ async function doFetch(
     if (signal.aborted) {
       return;
     }
+    console.error(`[doFetch] Error for request ${request.id}:`, err);
     throw err;
   }
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@deco-cx/warp-node",
-  "version": "0.3.20",
+  "version": "0.3.21-beta.1",
   "description": "WebSocket tunneling library for Node.js and other runtimes",
   "type": "module",
   "main": "dist/mod.js",