-
Notifications
You must be signed in to change notification settings - Fork 392
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
CB-6000 handle task info events from WS
- Loading branch information
Showing
24 changed files
with
254 additions
and
150 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
23 changes: 23 additions & 0 deletions
23
webapp/packages/core-root/src/AsyncTask/AsyncTaskInfoEventHandler.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| /* | ||
| * CloudBeaver - Cloud Database Manager | ||
| * Copyright (C) 2020-2024 DBeaver Corp and others | ||
| * | ||
| * Licensed under the Apache License, Version 2.0. | ||
| * you may not use this file except in compliance with the License. | ||
| */ | ||
| import { injectable } from '@cloudbeaver/core-di'; | ||
| import type { WsAsyncTaskInfo } from '@cloudbeaver/core-sdk'; | ||
|
|
||
| import { TopicEventHandler } from '../ServerEventEmitter/TopicEventHandler.js'; | ||
| import { type ISessionEvent, type SessionEventId, SessionEventSource, SessionEventTopic } from '../SessionEventSource.js'; | ||
|
|
||
| @injectable() | ||
| export class AsyncTaskInfoEventHandler extends TopicEventHandler<WsAsyncTaskInfo, ISessionEvent, SessionEventId, SessionEventTopic> { | ||
| constructor(sessionEventSource: SessionEventSource) { | ||
| super(SessionEventTopic.CbSessionTask, sessionEventSource); | ||
| } | ||
|
|
||
| map(event: any): WsAsyncTaskInfo { | ||
| return event; | ||
| } | ||
| } |
157 changes: 157 additions & 0 deletions
157
webapp/packages/core-root/src/AsyncTask/AsyncTaskInfoService.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| /* | ||
| * CloudBeaver - Cloud Database Manager | ||
| * Copyright (C) 2020-2024 DBeaver Corp and others | ||
| * | ||
| * Licensed under the Apache License, Version 2.0. | ||
| * you may not use this file except in compliance with the License. | ||
| */ | ||
| import type { Subscription } from 'rxjs'; | ||
|
|
||
| import { Disposable, injectable } from '@cloudbeaver/core-di'; | ||
| import { type AsyncTaskInfo, GraphQLService, type WsAsyncTaskInfo } from '@cloudbeaver/core-sdk'; | ||
|
|
||
| import type { Unsubscribe } from '../ServerEventEmitter/IServerEventEmitter.js'; | ||
| import { ServerEventId } from '../SessionEventSource.js'; | ||
| import { AsyncTask } from './AsyncTask.js'; | ||
| import { AsyncTaskInfoEventHandler } from './AsyncTaskInfoEventHandler.js'; | ||
|
|
||
| @injectable() | ||
| export class AsyncTaskInfoService extends Disposable { | ||
| private readonly tasks: Map<string, AsyncTask>; | ||
| private readonly taskIdAliases: Map<string, string>; | ||
| private readonly pendingEvents: Map<string, WsAsyncTaskInfo>; | ||
| private connection: Subscription | null; | ||
| private onEventUnsubscribe: Unsubscribe | null; | ||
|
|
||
| constructor( | ||
| private readonly graphQLService: GraphQLService, | ||
| private readonly asyncTaskInfoEventHandler: AsyncTaskInfoEventHandler, | ||
| ) { | ||
| super(); | ||
| this.tasks = new Map(); | ||
| this.taskIdAliases = new Map(); | ||
| this.pendingEvents = new Map(); | ||
| this.connection = null; | ||
| this.handleEvent = this.handleEvent.bind(this); | ||
|
|
||
| this.onEventUnsubscribe = asyncTaskInfoEventHandler.onEvent<WsAsyncTaskInfo>(ServerEventId.CbSessionTaskInfoUpdated, this.handleEvent); | ||
| } | ||
|
|
||
| private async updateTask(task: AsyncTask, data: WsAsyncTaskInfo) { | ||
| if (data.running === false) { | ||
| await task.updateInfoAsync(async () => { | ||
| const { taskInfo } = await this.graphQLService.sdk.getAsyncTaskInfo({ | ||
| taskId: data.taskId, | ||
| removeOnFinish: false, | ||
| }); | ||
|
|
||
| return taskInfo; | ||
| }); | ||
| } else { | ||
| task.updateStatus(data); | ||
| } | ||
| } | ||
|
|
||
| private async handleEvent(data: WsAsyncTaskInfo) { | ||
| const task = this.getTask(data.taskId); | ||
|
|
||
| if (!task) { | ||
| this.pendingEvents.set(data.taskId, data); | ||
| return; | ||
| } | ||
|
|
||
| await this.updateTask(task, data); | ||
| } | ||
|
|
||
| override dispose(): void { | ||
| this.connection?.unsubscribe(); | ||
| this.onEventUnsubscribe?.(); | ||
| } | ||
|
|
||
| create(getter: () => Promise<AsyncTaskInfo>): AsyncTask { | ||
| const task = new AsyncTask(getter, this.cancelTask.bind(this)); | ||
|
|
||
| this.tasks.set(task.id, task); | ||
| task.onStatusChange.addHandler(info => { | ||
| if (this.taskIdAliases.get(info.id)) { | ||
| return; | ||
| } | ||
|
|
||
| this.taskIdAliases.set(info.id, task.id); | ||
|
|
||
| const pendingEvent = this.pendingEvents.get(info.id); | ||
| if (pendingEvent) { | ||
| this.pendingEvents.delete(info.id); | ||
| this.updateTask(task, pendingEvent); | ||
| } | ||
| }); | ||
|
|
||
| if (this.tasks.size === 1) { | ||
| this.connection = this.asyncTaskInfoEventHandler.eventsSubject.connect(); | ||
| } | ||
|
|
||
| return task; | ||
| } | ||
|
|
||
| private getTask(taskId: string): AsyncTask | undefined { | ||
| let task = this.tasks.get(taskId); | ||
|
|
||
| if (!task) { | ||
| const internalId = this.taskIdAliases.get(taskId); | ||
|
|
||
| if (internalId) { | ||
| task = this.tasks.get(internalId); | ||
| } | ||
| } | ||
|
|
||
| return task; | ||
| } | ||
|
|
||
| async run(task: AsyncTask): Promise<AsyncTaskInfo> { | ||
| if (task.info === null) { | ||
| await task.run(); | ||
| } | ||
|
|
||
| return task.promise; | ||
| } | ||
|
|
||
| async remove(taskId: string): Promise<void> { | ||
| const task = this.getTask(taskId); | ||
|
|
||
| if (!task) { | ||
| return; | ||
| } | ||
|
|
||
| if (task.pending) { | ||
| throw new Error('Cant remove unfinished task'); | ||
| } | ||
| this.tasks.delete(task.id); | ||
| if (task.info) { | ||
| this.taskIdAliases.delete(task.info.id); | ||
| this.pendingEvents.delete(task.info.id); | ||
| } | ||
| if (this.tasks.size === 0) { | ||
| this.connection?.unsubscribe(); | ||
| this.connection = null; | ||
| } | ||
|
|
||
| if (task.info !== null) { | ||
| await this.graphQLService.sdk.getAsyncTaskInfo({ | ||
| taskId: task.info.id, | ||
| removeOnFinish: true, | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| async cancel(taskId: string): Promise<void> { | ||
| const task = this.getTask(taskId); | ||
|
|
||
| await task?.cancelAsync(); | ||
| } | ||
|
|
||
| private async cancelTask(id: string): Promise<void> { | ||
| await this.graphQLService.sdk.asyncTaskCancel({ | ||
| taskId: id, | ||
| }); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.