ftp works but it blocks the thread

client/javascript/session.js

@@ -29,7 +29,6 @@ class CurlSession {
     let end_callback = (error) => {
       Module.removeFunction(end_callback_ptr);
       Module.removeFunction(data_callback_ptr);
-      Module.removeFunction(headers_callback_ptr);
       
       this.active_requests--;
       js_end_callback(error);
@@ -41,12 +40,14 @@ class CurlSession {
       js_data_callback(chunk);
     }
   
-    let headers_callback = () => {
-      js_headers_callback();
+    let headers_callback = (chunk_ptr, chunk_size) => {
+      let data = Module.HEAPU8.subarray(chunk_ptr, chunk_ptr + chunk_size);
+      let chunk = new Uint8Array(data);
+      js_headers_callback(chunk);
     }
   
     end_callback_ptr = Module.addFunction(end_callback, "vi");
-    headers_callback_ptr = Module.addFunction(headers_callback, "v");
+    headers_callback_ptr = Module.addFunction(headers_callback, "vii");
     data_callback_ptr = Module.addFunction(data_callback, "vii");
     let request_ptr = c_func(_create_request, [url, data_callback_ptr, end_callback_ptr, headers_callback_ptr]);
     
@@ -78,6 +79,7 @@ class CurlSession {
     this.event_loop = setInterval(() => {
       let libcurl_active = _session_get_active(this.session_ptr);
       if (libcurl_active || this.active_requests) {
+        console.log("test");
         _session_perform(this.session_ptr);
       }
       else {
@@ -95,4 +97,65 @@ class CurlSession {
     _session_cleanup(this.session_ptr);
     this.session_ptr = null;
   }
+
+  //wrap request callbacks using a readable stream and return the new callbacks
+  stream_response(url, headers_callback, end_callback, abort_signal) {
+    let stream_controller;
+    let aborted = false;
+    let headers_received = false;
+
+    let stream = new ReadableStream({
+      start(controller) {
+        stream_controller = controller;
+      }
+    });
+
+    if (abort_signal instanceof AbortSignal) {
+      abort_signal.addEventListener("abort", () => {
+        if (aborted) return;
+        aborted = true;
+        if (headers_received) {
+          stream_controller.error("The operation was aborted.");
+        }
+        real_abort_callback();
+      });
+    }
+
+    let real_data_callback = (new_data) => {
+      if (!headers_received) {
+        headers_received = true;
+        headers_callback(stream);
+      }
+
+      try {
+        stream_controller.enqueue(new_data);
+      }
+      catch (e) {
+        //the readable stream has been closed elsewhere, so cancel the request
+        if (e instanceof TypeError) {
+          finish_callback(-1);
+        }
+        else {
+          throw e;
+        }
+      }
+    }
+
+    let real_end_callback = (error) => {
+      if (!headers_received && error === 0) {
+        headers_received = true;
+        headers_callback(stream);
+      }
+
+      if (error != 0) {
+        try {
+          stream_controller.close();
+        }
+        catch {}
+      }
+      end_callback(error);
+    }
+
+    return this.create_request(url, real_data_callback, real_end_callback, () => {});
+  }
 }