Skip to content
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
68 changes: 68 additions & 0 deletions webapp/packages/core-root/src/AsyncTask/AsyncTask.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@ export class AsyncTask {
private readonly init: () => Promise<AsyncTaskInfo>;
private readonly cancel: (id: string) => Promise<void>;
private initPromise: Promise<void> | null;
private pollingTimer: ReturnType<typeof setInterval> | null;
private pollingGetter: ((taskId: string) => Promise<AsyncTaskInfo>) | null;

constructor(init: () => Promise<AsyncTaskInfo>, cancel: (id: string) => Promise<void>) {
this._id = uuid();
Expand All @@ -53,6 +55,8 @@ export class AsyncTask {
this.updatingAsync = false;
this.taskInfo = null;
this.initPromise = null;
this.pollingTimer = null;
this.pollingGetter = null;
this.onStatusChange = new SyncExecutor();

this.innerPromise = new Promise((resolve, reject) => {
Expand Down Expand Up @@ -118,6 +122,8 @@ export class AsyncTask {
}

this._cancelled = true;
// 停止轮询
this.stopPolling();
try {
await this.cancelTask();
} catch (exception: any) {
Expand All @@ -137,6 +143,8 @@ export class AsyncTask {
this.taskInfo = info;

if (!info.running) {
// 任务完成,停止轮询
this.stopPolling();
if (info.error) {
this.reject(new ServerInternalError(info.error));
} else {
Expand All @@ -151,4 +159,64 @@ export class AsyncTask {
await this.cancel(this.info.id);
}
}

/**
* 启动轮询机制,作为 WebSocket 通知的降级方案
* @param getter 用于获取任务信息的函数
* @param interval 轮询间隔(毫秒),默认 1000ms
*/
startPolling(getter: (taskId: string) => Promise<AsyncTaskInfo>, interval: number = 1000): void {
// 如果任务已经完成或已取消,不启动轮询
if (!this.pending || this._cancelled) {
return;
}

// 如果已经有轮询在运行,先停止
this.stopPolling();

this.pollingGetter = getter;

// 立即执行一次检查
this.pollOnce();

// 设置定时轮询
this.pollingTimer = setInterval(() => {
this.pollOnce();
}, interval);
}

/**
* 停止轮询
*/
stopPolling(): void {
if (this.pollingTimer) {
clearInterval(this.pollingTimer);
this.pollingTimer = null;
}
this.pollingGetter = null;
}

/**
* 执行一次轮询检查
*/
private async pollOnce(): Promise<void> {
// 如果任务已完成、已取消或没有轮询 getter,不执行
if (!this.pending || this._cancelled || !this.pollingGetter || !this.taskInfo) {
return;
}

// 如果正在更新,跳过本次轮询
if (this.updatingAsync) {
return;
}

try {
const info = await this.pollingGetter(this.taskInfo.id);
await this.updateInfoAsync(() => Promise.resolve(info));
} catch {
// 轮询失败不影响主流程,静默处理
// 如果 WebSocket 正常工作,会通过 WebSocket 通知完成任务
// 如果 WebSocket 连接中断,下次轮询会继续尝试
}
}
}
19 changes: 19 additions & 0 deletions webapp/packages/core-root/src/AsyncTask/AsyncTaskInfoService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,21 @@ export class AsyncTaskInfoService extends Disposable {
await task.run();
}

// 如果任务还在运行,启动轮询作为 WebSocket 的降级方案
// 这样可以确保即使 WebSocket 连接中断,也能通过轮询获取结果
if (task.pending && task.info) {
task.startPolling(
async (taskId: string) => {
const { taskInfo } = await this.graphQLService.sdk.getAsyncTaskInfo({
taskId,
removeOnFinish: false,
});
return taskInfo;
},
1000, // 轮询间隔 1 秒
);
}

return task.promise;
}

Expand All @@ -125,6 +140,10 @@ export class AsyncTaskInfoService extends Disposable {
if (task.pending) {
throw new Error('Cant remove unfinished task');
}

// 停止轮询
task.stopPolling();

this.tasks.delete(task.id);
if (task.info) {
this.taskIdAliases.delete(task.info.id);
Expand Down
Loading