diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts index 4fb7439bf8..ef4b60cfd6 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts @@ -20,7 +20,9 @@ describe('MemoryTaskBroker', () => { it('should claim a dispatched work item', async () => { const broker = new MemoryTaskBroker(); - await broker.dispatch({}); + await broker.dispatch({ + metadata: '', + }); await expect(broker.claim()).resolves.toEqual(expect.any(TaskAgent)); }); @@ -31,16 +33,16 @@ describe('MemoryTaskBroker', () => { await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting'); - await broker.dispatch({}); + await broker.dispatch({ metadata: 'foo' }); await expect(promise).resolves.toEqual(expect.any(TaskAgent)); }); it('should dispatch multiple items and claim them in order', async () => { const broker = new MemoryTaskBroker(); - await broker.dispatch({ name: 'a' }); - await broker.dispatch({ name: 'b' }); - await broker.dispatch({ name: 'c' }); + await broker.dispatch({ metadata: 'a' }); + await broker.dispatch({ metadata: 'b' }); + await broker.dispatch({ metadata: 'c' }); const taskA = await broker.claim(); const taskB = await broker.claim(); @@ -48,9 +50,9 @@ describe('MemoryTaskBroker', () => { await expect(taskA).toEqual(expect.any(TaskAgent)); await expect(taskB).toEqual(expect.any(TaskAgent)); await expect(taskC).toEqual(expect.any(TaskAgent)); - await expect(taskA.spec.name).toBe('a'); - await expect(taskB.spec.name).toBe('b'); - await expect(taskC.spec.name).toBe('c'); + await expect(taskA.spec.metadata).toBe('a'); + await expect(taskB.spec.metadata).toBe('b'); + await expect(taskC.spec.metadata).toBe('c'); }); it('should complete a task', async () => { diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts index 322e669373..c6fb56b42a 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts @@ -14,14 +14,7 @@ * limitations under the License. */ -import { - CompletedTaskState, - Task, - TaskSpec, - TaskBroker, - Status, -} from './types'; -import { v4 as uuid } from 'uuid'; +import { CompletedTaskState, Task, TaskSpec, TaskBroker } from './types'; import { InMemoryDatabase } from './database'; export class TaskAgent implements Task { @@ -48,20 +41,25 @@ export class TaskAgent implements Task { } async complete(result: CompletedTaskState): Promise { - this.state.status = result === 'FAILED' ? 'COMPLETED' : 'FAILED'; + this.db.setStatus( + this.state.taskId, + result === 'FAILED' ? 'COMPLETED' : 'FAILED', + ); } private start() { this.heartbeartInterval = setInterval(() => { - const runId = 'iiiid'; - this.db.heartBeat(runId); - }, 4269); + if (!this.state.runId) { + throw new Error('no run id provided'); + } + this.db.heartBeat(this.state.runId); + }, 1000); } } interface TaskState { spec: TaskSpec; - status: Status; + taskId: string; runId: string | undefined; } @@ -75,14 +73,20 @@ function defer() { export class MemoryTaskBroker implements TaskBroker { private readonly db = new InMemoryDatabase(); - private readonly tasks = new Array(); private deferredDispatch = defer(); async claim(): Promise { for (;;) { const pendingTask = await this.db.claimTask(); if (pendingTask) { - return TaskAgent.create(pendingTask, this.db); + return TaskAgent.create( + { + runId: pendingTask.runId, + taskId: pendingTask.taskId, + spec: pendingTask.spec, + }, + this.db, + ); } await this.waitForDispatch(); @@ -90,11 +94,7 @@ export class MemoryTaskBroker implements TaskBroker { } async dispatch(spec: TaskSpec): Promise { - this.tasks.push({ - spec, - status: 'OPEN', - runId: undefined, - }); + await this.db.createTask(spec); this.signalDispatch(); }