feat(client): request abort signal support

libs/client/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@fal-ai/client",
   "description": "The fal.ai client for JavaScript and TypeScript",
-  "version": "1.2.0-alpha.5",
+  "version": "1.2.0-alpha.6",
   "license": "MIT",
   "repository": {
     "type": "git",

libs/client/src/client.ts

@@ -106,6 +106,9 @@ export function createFalClient(userConfig: Config = {}): FalClient {
           ...config,
           responseHandler: resultResponseHandler,
         },
+        options: {
+          signal: options.abortSignal,
+        },
       });
     },
     subscribe: async (endpointId, options) => {

libs/client/src/queue.ts

@@ -123,6 +123,11 @@ type BaseQueueOptions = {
    * The unique identifier for the enqueued request.
    */
   requestId: string;
+
+  /**
+   * The signal to abort the request.
+   */
+  abortSignal?: AbortSignal;
 };
 
 export type QueueStatusOptions = BaseQueueOptions & {
@@ -246,11 +251,14 @@ export const createQueueClient = ({
         },
         input: input as Input,
         config,
+        options: {
+          signal: options.abortSignal,
+        },
       });
     },
     async status(
       endpointId: string,
-      { requestId, logs = false }: QueueStatusOptions,
+      { requestId, logs = false, abortSignal }: QueueStatusOptions,
     ): Promise<QueueStatus> {
       const appId = parseEndpointId(endpointId);
       const prefix = appId.namespace ? `${appId.namespace}/` : "";
@@ -262,6 +270,9 @@ export const createQueueClient = ({
           path: `/requests/${requestId}/status`,
         }),
         config,
+        options: {
+          signal: abortSignal,
+        },
       });
     },
 
@@ -379,6 +390,7 @@ export const createQueueClient = ({
             const requestStatus = await ref.status(endpointId, {
               requestId,
               logs: options.logs ?? false,
+              abortSignal: options.abortSignal,
             });
             if (options.onQueueUpdate) {
               options.onQueueUpdate(requestStatus);
@@ -400,7 +412,7 @@ export const createQueueClient = ({
 
     async result<Output>(
       endpointId: string,
-      { requestId }: BaseQueueOptions,
+      { requestId, abortSignal }: BaseQueueOptions,
     ): Promise<Result<Output>> {
       const appId = parseEndpointId(endpointId);
       const prefix = appId.namespace ? `${appId.namespace}/` : "";
@@ -414,12 +426,15 @@ export const createQueueClient = ({
           ...config,
           responseHandler: resultResponseHandler,
         },
+        options: {
+          signal: abortSignal,
+        },
       });
     },
 
     async cancel(
       endpointId: string,
-      { requestId }: BaseQueueOptions,
+      { requestId, abortSignal }: BaseQueueOptions,
     ): Promise<void> {
       const appId = parseEndpointId(endpointId);
       const prefix = appId.namespace ? `${appId.namespace}/` : "";
@@ -430,6 +445,9 @@ export const createQueueClient = ({
           path: `/requests/${requestId}/cancel`,
         }),
         config,
+        options: {
+          signal: abortSignal,
+        },
       });
     },
   };

libs/client/src/storage.ts

@@ -45,7 +45,7 @@
  * @returns the file extension or `bin` if the content type is not recognized.
  */
 function getExtensionFromContentType(contentType: string): string {
   const [_, fileType] = contentType.split("/");
   return fileType.split(/[-;]/)[0] ?? "bin";
 }
 
@@ -105,7 +105,7 @@
   uploadUrl: string,
   chunk: Blob,
   config: RequiredConfig,
-  tries: number = 3,
+  tries = 3,
 ): Promise<MultipartObject> {
   if (tries === 0) {
     throw new Error("Part upload failed, retries exhausted");
@@ -142,21 +142,17 @@
 
   const responses: MultipartObject[] = [];
 
-  try {
-    for (let i = 0; i < chunks; i++) {
-      const start = i * chunkSize;
-      const end = Math.min(start + chunkSize, file.size);
+  for (let i = 0; i < chunks; i++) {
+    const start = i * chunkSize;
+    const end = Math.min(start + chunkSize, file.size);
 
-      const chunk = file.slice(start, end);
+    const chunk = file.slice(start, end);
 
-      const partNumber = i + 1;
-      // {uploadUrl}/{part_number}?uploadUrlParams=...
-      const partUploadUrl = `${parsedUrl.origin}${parsedUrl.pathname}/${partNumber}${parsedUrl.search}`;
+    const partNumber = i + 1;
+    // {uploadUrl}/{part_number}?uploadUrlParams=...
+    const partUploadUrl = `${parsedUrl.origin}${parsedUrl.pathname}/${partNumber}${parsedUrl.search}`;
 
-      responses.push(await partUploadRetries(partUploadUrl, chunk, config));
-    }
-  } catch (error) {
-    throw error;
+    responses.push(await partUploadRetries(partUploadUrl, chunk, config));
   }
 
   // Complete the upload
@@ -221,7 +217,7 @@
       } else if (input instanceof Blob) {
         return await ref.upload(input);
       } else if (isPlainObject(input)) {
         const inputObject = input as Record<string, any>;
         const promises = Object.entries(inputObject).map(
           async ([key, value]): Promise<KeyValuePair> => {
             return [key, await ref.transformInput(value)];

libs/client/src/streaming.ts

@@ -63,13 +63,18 @@
    * support streaming.
    */
   readonly connectionMode?: StreamingConnectionMode;
+
+  /**
+   * The signal to abort the request.
+   */
+  readonly signal?: AbortSignal;
 };
 
 const EVENT_STREAM_TIMEOUT = 15 * 1000;
 
 type FalStreamEventType = "data" | "error" | "done";
 
 type EventHandler<T = any> = (event: T) => void;
 
 /**
  * The class representing a streaming response. With t
@@ -129,6 +134,14 @@
         reject(error);
       });
     });
+    // if a abort signal was passed, sync it with the internal one
+    if (options.signal) {
+      options.signal.addEventListener("abort", () => {
+        this.abortController.abort();
+      });
+    }
+
+    // start the streaming request
     this.start().catch(this.handleError);
   }
 
@@ -232,7 +245,7 @@
           this.emit("data", parsedData);
 
           // also emit 'message'for backwards compatibility
           this.emit("message" as any, parsedData);
         } catch (e) {
           this.emit("error", e);
         }
@@ -268,7 +281,7 @@
     return;
   };
 
   private handleError = (error: any) => {
     // In case AbortError is thrown but the signal is marked as aborted
     // it means the user called abort() and we should not emit an error
     // as it's expected behavior
@@ -345,6 +358,10 @@
 
   /**
    * Gets the `AbortSignal` instance that can be used to listen for abort events.
+   *
+   * **Note:** this signal is internal to the `FalStream` instance. If you pass your
+   * own abort signal, the `FalStream` will listen to it and abort it appropriately.
+   *
    * @returns the `AbortSignal` instance.
    * @see https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal
    */

libs/client/src/types/common.ts

@@ -22,6 +22,11 @@ export type RunOptions<Input> = {
    * The HTTP method, defaults to `post`;
    */
   readonly method?: "get" | "post" | "put" | "delete" | string;
+
+  /**
+   * The abort signal to cancel the request.
+   */
+  readonly abortSignal?: AbortSignal;
 };
 
 export type UrlOptions = {