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);

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<PipelineTask>,
@@ -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`];
   }