Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(client): queue path support #20

Merged
merged 1 commit into from
Oct 27, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion libs/client/package.json
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"name": "@fal-ai/serverless-client",
"description": "The fal serverless JS/TS client",
"version": "0.3.2",
"version": "0.4.0",
"license": "MIT",
"repository": {
"type": "git",
Expand Down
61 changes: 48 additions & 13 deletions libs/client/src/function.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import { isUUIDv4, isValidUrl } from './utils';
*/
type RunOptions<Input> = {
/**
* The path to the function, if any. Defaults to `/`.
* The path to the function, if any. Defaults to ``.
*/
readonly path?: string;

Expand Down Expand Up @@ -126,19 +126,24 @@ export async function subscribe<Input, Output>(
if (options.onEnqueue) {
options.onEnqueue(requestId);
}
const path = options.path ?? '';
return new Promise<Output>((resolve, reject) => {
let timeoutId: ReturnType<typeof setTimeout>;
const pollInterval = options.pollInterval ?? 1000;
const poll = async () => {
try {
const requestStatus = await queue.status(id, requestId, options.logs ?? false);
const requestStatus = await queue.status(id, {
requestId,
logs: options.logs ?? false,
path,
});
if (options.onQueueUpdate) {
options.onQueueUpdate(requestStatus);
}
if (requestStatus.status === 'COMPLETED') {
clearTimeout(timeoutId);
try {
const result = await queue.result<Output>(id, requestId);
const result = await queue.result<Output>(id, { requestId, path });
resolve(result);
} catch (error) {
reject(error);
Expand Down Expand Up @@ -179,6 +184,26 @@ type QueueSubscribeOptions = {

/**
* If `true`, the response will include the logs for the request.
* Defaults to `false`.
*/
logs?: boolean;
};

type BaseQueueOptions = {
/**
* The unique identifier for the enqueued request.
*/
requestId: string;
/**
* The path to the function, if any. Defaults to ``.
*/
path?: string;
};

type QueueStatusOptions = BaseQueueOptions & {
/**
* If `true`, the response will include the logs for the request.
* Defaults to `false`.
*/
logs?: boolean;
};
Expand All @@ -201,20 +226,19 @@ interface Queue {
* Retrieves the status of a specific request in the queue.
*
* @param id - The ID or URL of the function web endpoint.
* @param requestId - The unique identifier for the enqueued request.
* @param logs - If `true`, the response will include the logs for the request.
* @param options - Options to configure how the request is run.
* @returns A promise that resolves to the status of the request.
*/
status(id: string, requestId: string, logs: boolean): Promise<QueueStatus>;
status(id: string, options: QueueStatusOptions): Promise<QueueStatus>;

/**
* Retrieves the result of a specific request from the queue.
*
* @param id - The ID or URL of the function web endpoint.
* @param requestId - The unique identifier for the enqueued request.
* @param options - Options to configure how the request is run.
* @returns A promise that resolves to the result of the request.
*/
result<Output>(id: string, requestId: string): Promise<Output>;
result<Output>(id: string, options: BaseQueueOptions): Promise<Output>;

/**
* @deprecated Use `fal.subscribe` instead.
Expand All @@ -235,21 +259,32 @@ export const queue: Queue = {
id: string,
options: RunOptions<Input>
): Promise<EnqueueResult> {
return run(id, { ...options, method: 'post', path: '/fal/queue/submit/' });
const path = options.path ?? '';
return run(id, {
...options,
method: 'post',
path: '/fal/queue/submit' + path,
});
},
async status(id: string, requestId: string, logs = false): Promise<QueueStatus> {
async status(
id: string,
{ requestId, logs = false, path = '' }: QueueStatusOptions
): Promise<QueueStatus> {
return run(id, {
method: 'get',
path: `/fal/queue/requests/${requestId}/status`,
path: `/fal/queue/requests/${requestId}/status${path}`,
input: {
logs: logs ? '1' : '0',
},
});
},
async result<Output>(id: string, requestId: string): Promise<Output> {
async result<Output>(
id: string,
{ requestId, path = '' }: BaseQueueOptions
): Promise<Output> {
return run(id, {
method: 'get',
path: `/fal/queue/requests/${requestId}/response`,
path: `/fal/queue/requests/${requestId}/response${path}`,
});
},
subscribe,
Expand Down
Loading