diff --git a/src/pipeline-tasks/pipeline-tasks.resolver.ts b/src/pipeline-tasks/pipeline-tasks.resolver.ts index b3e3d77..603f39e 100644 --- a/src/pipeline-tasks/pipeline-tasks.resolver.ts +++ b/src/pipeline-tasks/pipeline-tasks.resolver.ts @@ -29,6 +29,15 @@ export class PipelineTasksResolver { return asyncIterator; } + @Subscription(() => PipelineTask, { + resolve: (value) => { + return value; + }, + }) + async pipelineTaskChanged(@Args('id') id: string) { + return await this.service.watchTaskUpdated(id); + } + @Query(() => [PipelineTask]) async listPipelineTaskByPipelineId(@Args('pipelineId') pipelineId: string) { return await this.service.listTasksByPipelineId(pipelineId); diff --git a/src/pipeline-tasks/pipeline-tasks.service.ts b/src/pipeline-tasks/pipeline-tasks.service.ts index 8258aad..f01cc51 100644 --- a/src/pipeline-tasks/pipeline-tasks.service.ts +++ b/src/pipeline-tasks/pipeline-tasks.service.ts @@ -9,9 +9,11 @@ import { InjectQueue } from '@nestjs/bull'; import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants'; import { Queue } from 'bull'; import { LockFailedException } from '../commons/exceptions/lock-failed.exception'; +import { PubSub } from 'apollo-server-express'; @Injectable() export class PipelineTasksService { + pubSub = new PubSub(); constructor( @InjectRepository(PipelineTask) private readonly repository: Repository, @@ -80,9 +82,14 @@ export class PipelineTasksService { } async updateTask(task: PipelineTask) { + this.pubSub.publish(task.id, task); return await this.repository.save(task); } + async watchTaskUpdated(id: string) { + return this.pubSub.asyncIterator(id); + } + getRedisTokens(pipeline: Pipeline): [string, string] { return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`]; }