Fix/warp handler message (#5)

* Fix warp message closed Signed-off-by: Marcos Candeia <[email protected]> * Fix server stream cancel Signed-off-by: Marcos Candeia <[email protected]> * Add error description Signed-off-by: Marcos Candeia <[email protected]> * Ignore abort error Signed-off-by: Marcos Candeia <[email protected]> * Close chan Signed-off-by: Marcos Candeia <[email protected]> * Check if target is aborted Signed-off-by: Marcos Candeia <[email protected]> * Add signal to makeReadableStream Signed-off-by: Marcos Candeia <[email protected]> * Controller error when signal is aborted Signed-off-by: Marcos Candeia <[email protected]> * Add controller comment to allow error Signed-off-by: Marcos Candeia <[email protected]> * bump version Signed-off-by: Marcos Candeia <[email protected]> --------- Signed-off-by: Marcos Candeia <[email protected]>
deco-cx · Aug 22, 2024 · 262b692 · 262b692
1 parent a6023d0
commit 262b692
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 6 deletions.
diff --git a/channel.ts b/channel.ts
@@ -167,12 +167,17 @@ export const makeWebSocket = <
 
 export const makeReadableStream = (
   ch: Channel<Uint8Array>,
+  signal?: AbortSignal,
 ): ReadableStream<Uint8Array> => {
   return new ReadableStream({
     async start(controller) {
-      for await (const content of ch.recv()) {
+      for await (const content of ch.recv(signal)) {
         controller.enqueue(content);
       }
+      // Uncomment if necessary. this will send a signal to the controller
+      // if (signal?.aborted) {
+      //   controller.error(new Error("aborted"));
+      // }
       controller.close();
     },
     cancel() {
@@ -195,6 +200,10 @@ export const makeChanStream = (
     }
     chan.close();
   };
-  processStream().catch(console.error);
+  processStream().catch((err) => {
+    if (!err?.target?.aborted) {
+      console.error("error processing stream", err);
+    }
+  });
   return chan;
 };
diff --git a/deno.json b/deno.json
@@ -1,6 +1,6 @@
 {
   "name": "@deco/warp",
-  "version": "0.3.5",
+  "version": "0.3.6",
   "exports": "./mod.ts",
   "tasks": {
     "check": "deno fmt && deno lint && deno check mod.ts"

diff --git a/handlers.client.ts b/handlers.client.ts
@@ -56,7 +56,9 @@ const onRequestStart: ServerMessageHandler<RequestStartMessage> = async (
   if (!message.hasBody) {
     doFetch(message, state, state.ch.out, abortCtrl.signal).catch(
       ignoreIfClosed,
-    );
+    ).finally(() => {
+      delete state.requests[message.id];
+    });
   } else {
     const bodyData = makeChan<Uint8Array>();
     state.requests[message.id]!.body = bodyData;

diff --git a/handlers.server.ts b/handlers.server.ts
@@ -48,7 +48,7 @@ const onResponseStart: ClientMessageHandler<ResponseStartMessage> = (
   );
   const shouldBeNullBody = NULL_BODIES.includes(message.statusCode);
   const stream = !shouldBeNullBody && request.responseBodyChan
-    ? makeReadableStream(request.responseBodyChan)
+    ? makeReadableStream(request.responseBodyChan, state.ch.out.signal)
     : undefined;
   const resp = new Response(stream, {
     status: message.statusCode,
@@ -62,6 +62,7 @@ const onResponseStart: ClientMessageHandler<ResponseStartMessage> = (
       request.responseObject.reject(
         new DOMException("Connection closed", "AbortError"),
       );
+      request?.responseBodyChan?.close?.();
     }
   });
   request.responseObject.resolve(resp);
@@ -105,7 +106,7 @@ const onDataEnd: ClientMessageHandler<DataEndMessage> = (state, message) => {
   try {
     request.responseBodyChan?.close?.();
   } catch (_err) {
-    console.log(_err);
+    console.log(`error closing body chan`, _err);
   }
 };