feat(pipelines): list commit logs 使用直接订阅。

替代了先查询再订阅，确保数据不丢失。

src/pipelines/pipelines.resolver.ts

@@ -1,4 +1,4 @@
-import { Args, Mutation, Query, Resolver } from '@nestjs/graphql';
+import { Args, Mutation, Query, Resolver, Subscription } from '@nestjs/graphql';
 import { CreatePipelineInput } from './dtos/create-pipeline.input';
 import { UpdatePipelineInput } from './dtos/update-pipeline.input';
 import { Pipeline } from './pipeline.entity';
@@ -43,8 +43,15 @@ export class PipelinesResolver {
     return await this.service.remove(id);
   }
 
-  @Query(() => String)
+  @Subscription(() => LogList, {
+    resolve: (value) => {
+      return value;
+    },
+  })
   async listLogsForPipeline(@Args('id', { type: () => String }) id: string) {
-    return await this.service.listLogsForPipeline(id);
+    const job = await this.service.listLogsForPipeline(id);
+    return (async function* () {
+      yield await job.finished();
+    })();
   }
 }

src/pipelines/pipelines.service.spec.ts

@@ -59,7 +59,7 @@ describe('PipelinesService', () => {
       const add = jest.spyOn(queue, 'add');
       await expect(
         service.listLogsForPipeline('test-pipeline'),
-      ).resolves.toEqual(1);
+      ).resolves.toEqual({ id: 1 });
       expect(add).toBeCalledWith({
         project: pipeline.project,
         branch: pipeline.branch,

src/pipelines/pipelines.service.ts

@@ -50,6 +50,6 @@ export class PipelinesService extends BaseDbService<Pipeline> {
       project: pipeline.project,
       branch: pipeline.branch,
     });
-    return job.id;
+    return job;
   }
 }

src/repos/list-logs.consumer.ts

@@ -1,24 +1,14 @@
-import { PubSub } from 'graphql-subscriptions';
 import { ReposService } from './repos.service';
 import { Processor, Process } from '@nestjs/bull';
 import { Job } from 'bull';
 import { ListLogsOption } from './models/list-logs.options';
-import {
+import { LIST_LOGS_TASK } from './repos.constants';
-  LIST_LOGS_DONE,
-  LIST_LOGS_PUB_SUB,
-  LIST_LOGS_TASK,
-} from './repos.constants';
-import { Inject } from '@nestjs/common';
 @Processor(LIST_LOGS_TASK)
 export class ListLogsConsumer {
-  constructor(
+  constructor(private readonly service: ReposService) {}
-    private readonly service: ReposService,
-    @Inject(LIST_LOGS_PUB_SUB)
-    private readonly pubSub: PubSub,
-  ) {}
   @Process()
   async listLogs(job: Job<ListLogsOption>) {
     const logs = await this.service.listLogs(job.data);
-    this.pubSub.publish(LIST_LOGS_DONE, logs);
+    return logs;
   }
 }

src/repos/repos.module.ts

@@ -19,15 +19,7 @@ import { ListLogsConsumer } from './list-logs.consumer';
       name: LIST_LOGS_TASK,
     }),
   ],
-  providers: [
+  providers: [ReposResolver, ReposService, ListLogsConsumer],
-    ReposResolver,
-    ReposService,
-    ListLogsConsumer,
-    {
-      provide: LIST_LOGS_PUB_SUB,
-      useValue: new PubSub(),
-    },
-  ],
   exports: [ReposService],
 })
 export class ReposModule {}

src/repos/repos.resolver.ts

@@ -1,21 +1,4 @@
-import { LIST_LOGS_DONE, LIST_LOGS_PUB_SUB } from './repos.constants';
+import { Resolver } from '@nestjs/graphql';
-import { Resolver, Subscription } from '@nestjs/graphql';
-import { LogList } from './dtos/log-list.model';
-import { Inject } from '@nestjs/common';
-import { PubSub } from 'apollo-server-express';
 
 @Resolver()
-export class ReposResolver {
+export class ReposResolver {}
-  constructor(
-    @Inject(LIST_LOGS_PUB_SUB)
-    private readonly pubSub: PubSub,
-  ) {}
-  @Subscription(() => LogList, {
-    resolve: (value) => {
-      return value;
-    },
-  })
-  listLogsDone() {
-    return this.pubSub.asyncIterator(LIST_LOGS_DONE);
-  }
-}