WIP - ReadableStream for multipart

Author: lucas-amiaud

src/lib/client/MultipartHttpClient.ts

@@ -1,98 +1,136 @@
 import { HttpMethod } from 'simple-http-request-builder';
-import { HttpPromise, unwrapHttpPromise } from '../promise/HttpPromise';
-import { networkErrorCatcher } from './FetchClient';
 import {
-  genericError, HttpResponse, networkError, timeoutError,
-} from './HttpResponse';
+  FetchResponseHandler,
+  genericError,
+  HttpPromise,
+  HttpResponse,
+  networkErrorCatcher,
+  toErrorResponsePromise,
+  unwrapHttpPromise,
+} from 'simple-http-rest-client';
+import { Logger } from 'simple-logging-system';
 import {
   MultipartHttpClient,
   MultipartHttpOptions,
   MultipartHttpRequest,
 } from './MultipartHttpRequest';
 
-/**
- * Handle multipart request using {@link XMLHttpRequest}
- * @param multipartHttpRequest the request to be executed
- */
+const logger: Logger = new Logger('MultipartHttpClient');
+
 export const multipartHttpFetchClientExecutor: MultipartHttpClient<Promise<unknown>> = (
   multipartHttpRequest: MultipartHttpRequest<unknown>,
-): Promise<unknown> => {
-  const xhr: XMLHttpRequest = new XMLHttpRequest();
-
-  // Abort request after configured timeout time
-  const timeoutHandle: ReturnType<typeof setTimeout> = setTimeout(
-    () => xhr.abort(),
-    multipartHttpRequest.optionValues.timeoutInMillis,
-  );
-
-  // Return a promise that resolves when the request is complete
-  return new Promise<unknown>((resolve: (value: unknown) => void) => {
-    xhr.open(multipartHttpRequest.method, multipartHttpRequest.buildUrl(), true);
-
-    // Set credentials
-    xhr.withCredentials = multipartHttpRequest.optionValues.withCredentials;
-
-    // Set headers
-    if (multipartHttpRequest.headersValue) {
-      for (const [key, value] of Object.entries(multipartHttpRequest.headersValue)) {
-        xhr.setRequestHeader(key, value);
-      }
-    }
+): Promise<Response> => {
+  const { boundary }: MultipartHttpOptions = multipartHttpRequest.optionValues;
+  const stream: ReadableStream<unknown> = new ReadableStream({
+    start(controller: ReadableStreamDefaultController<unknown>) {
+      const encoder: TextEncoder = new TextEncoder();
+      const entries = Array.from(multipartHttpRequest.formData.entries());
+      const processNextEntry = (index: number) => {
+        if (index >= entries.length) {
+          // Finalize the stream with the closing boundary
+          controller.enqueue(encoder.encode(`--${boundary}--\r\n`));
+          controller.close();
+          return;
+        }
 
-    // Handle response
-    xhr.onload = () => {
-      if (xhr.status >= 200 && xhr.status < 300) {
-        return resolve({ response: JSON.parse(xhr.response) });
-      }
-      return resolve({ error: JSON.parse(xhr.response) ?? genericError });
-    };
+        const [name, data] = entries[index];
+        const enqueueBoundaryAndHeaders = () => {
+          controller.enqueue(encoder.encode(`--${boundary}\r\n`));
+          if (data instanceof Blob) {
+            // Binary data
+            const contentType = data.type || 'application/octet-stream';
+            controller.enqueue(encoder.encode(
+              `Content-Disposition: form-data; name="${name}"; filename="${name}"\r\n`
+              + `Content-Type: ${contentType}\r\n\r\n`,
+            ));
+            return data.stream().getReader();
+          }
+          // Text data
+          controller.enqueue(encoder.encode(
+            `Content-Disposition: form-data; name="${name}"\r\n\r\n`,
+          ));
+          controller.enqueue(encoder.encode(`${data.toString()}\r\n`));
+          return null;
+        };
 
-    // Handle network errors
-    xhr.onerror = () => resolve({ error: networkError });
+        const reader = enqueueBoundaryAndHeaders();
 
-    // Handle request timeout
-    xhr.ontimeout = () => resolve({ error: timeoutError });
+        // Handle text directly, or start processing binary data
+        if (!reader) {
+          processNextEntry(index + 1); // Process the next entry
+          return;
+        }
 
-    // Handle progress
-    xhr.upload.onprogress = (event: ProgressEvent) => {
-      multipartHttpRequest.optionValues.onProgressCallback(event);
-    };
+        // Process the binary data with recursion
+        const pumpReader = () => {
+          reader.read().then(({ done, value }) => {
+            if (done) {
+              // Move to the next entry
+              controller.enqueue(encoder.encode('\r\n'));
+              processNextEntry(index + 1);
+            } else {
+              // Enqueue the current chunk
+              controller.enqueue(value);
+              pumpReader(); // Process the next chunk
+            }
+          });
+        };
 
-    xhr.upload.onerror = () => resolve({ error: genericError });
+        pumpReader();
+      };
 
-    // Send the request
-    xhr.send(multipartHttpRequest.formData);
-  })
+      processNextEntry(0); // Start processing the first entry
+    },
+    cancel: (reason: unknown) => {
+      multipartHttpRequest.optionValues.timeoutAbortController.abort(reason);
+    },
+  });
+  const timeoutHandle = setTimeout(
+    () => stream.cancel('timeout'),
+    multipartHttpRequest.optionValues.timeoutInMillis,
+  );
+  return fetch(
+    multipartHttpRequest.buildUrl(),
+    {
+      headers: {
+        ...multipartHttpRequest.headersValue,
+        'Content-Type': `multipart/form-data; boundary=${boundary}`,
+      },
+      method: HttpMethod.POST,
+      body: stream,
+      signal: multipartHttpRequest.optionValues.timeoutAbortController.signal,
+      duplex: 'half',
+    },
+  )
     .finally(() => clearTimeout(timeoutHandle));
 };
 
-/**
- * A {@link MultipartHttpFetchClient} that executes an {@link MultipartHttpRequest} that returns JSON responses.
- * It uses {@link multipartHttpFetchClient} to executes the {@link MultipartHttpRequest}.
- */
 export const multipartHttpFetchClient = <T = void>(
   httpRequest: MultipartHttpRequest<unknown>,
+  ...handlers: FetchResponseHandler[]
 ): Promise<HttpResponse<T>> => <Promise<HttpResponse<T>>>multipartHttpFetchClientExecutor(httpRequest)
+  .then((response) => {
+    for (const handler of handlers) {
+      try {
+        const handlerResult = handler(response);
+        if (handlerResult !== undefined) {
+          return handlerResult;
+        }
+      } catch (error) {
+        logger.error('Error executing handler', { error });
+        return toErrorResponsePromise(genericError);
+      }
+    }
+    return { response };
+  })
   .catch(networkErrorCatcher);
 
 export type MultipartHttpFetchClient = <T>(
   multipartHttpRequest: MultipartHttpRequest<unknown>,
 ) => Promise<HttpResponse<T>>;
 
-/**
- * Factory function to create fetch {@link MultipartHttpRequest}.
- *
- * @param baseUrl The base URL. It should not contain an ending slash. A valid base URL is: http://hostname/api
- * @param method The HTTP method used for the request, see {@link HttpMethod}
- * @param path The path of the endpoint to call, it should be composed with a leading slash
- * and will be appended to the {@link MultipartHttpRequest#baseUrl}. A valid path is: /users
- * @param multipartHttpClient The fetch client that uses {@link MultipartHttpRequest} and returns
- * a `Promise<HttpResponse<T>>`
- * @param options Optional options to configure the request
- */
 export function createMultipartHttpFetchRequest<T>(
   baseUrl: string,
-  method: HttpMethod,
   path: string,
   multipartHttpClient: MultipartHttpFetchClient,
   options?: Partial<MultipartHttpOptions>,
@@ -103,7 +141,6 @@ export function createMultipartHttpFetchRequest<T>(
       multipartHttpRequest,
     ),
     baseUrl,
-    method,
     path,
     options,
   );

src/lib/client/MultipartHttpRequest.ts

@@ -1,22 +1,19 @@
-import { HttpMethod } from 'simple-http-request-builder';
-
 export type MultipartHttpOptions = {
   timeoutInMillis: number,
+  timeoutAbortController: AbortController,
   onProgressCallback: (event: ProgressEvent) => void,
-  withCredentials: boolean,
+  boundary: string,
 };
 
-export type MultipartHttpClient<T> = (request: MultipartHttpRequest<unknown>) => T;
+export interface MultipartHttpClient<T> {
+  (request: MultipartHttpRequest<unknown>): T;
+}
 
 export class MultipartHttpRequest<T> {
-  private static readonly DEFAULT_TIMEOUT_IN_MILLIS: number = 60_000; // 1 minute
-
   readonly multipartHttpClient: MultipartHttpClient<T>;
 
   readonly baseUrl: URL;
 
-  readonly method: HttpMethod;
-
   readonly path: string;
 
   readonly headersValue: HeadersInit;
@@ -28,20 +25,20 @@ export class MultipartHttpRequest<T> {
   constructor(
     multipartHttpClient: MultipartHttpClient<T>,
     baseUrl: string,
-    method: HttpMethod,
     path: string,
     options?: Partial<MultipartHttpOptions>,
   ) {
     this.multipartHttpClient = multipartHttpClient;
     this.baseUrl = new URL(baseUrl);
-    this.method = method;
     this.path = path;
     this.headersValue = {};
     this.formData = new FormData();
     this.optionValues = {
-      timeoutInMillis: options?.timeoutInMillis ?? MultipartHttpRequest.DEFAULT_TIMEOUT_IN_MILLIS,
-      onProgressCallback: options?.onProgressCallback ?? (() => {}),
-      withCredentials: options?.withCredentials ?? false,
+      timeoutInMillis: options?.timeoutInMillis ?? 60000,
+      timeoutAbortController: options?.timeoutAbortController ?? new AbortController(),
+      onProgressCallback: options?.onProgressCallback ?? (() => {
+      }),
+      boundary: options?.boundary ?? `boundary-${Date.now()}`,
     };
   }
 
@@ -59,18 +56,6 @@ export class MultipartHttpRequest<T> {
     return this;
   }
 
-  file(file: File) {
-    this.data([['file', file]]);
-    return this;
-  }
-
-  files(files: File[]) {
-    for (const file of files) {
-      this.file(file);
-    }
-    return this;
-  }
-
   buildUrl() {
     return encodeURI(this.baseUrl.toString() + this.path);
   }