Merge branch 'feat-pub-sub' of ssh://git.ivanli.cc:7018/Fennec/fennec-be into feat-pub-sub

package-lock.json

@@ -13,6 +13,7 @@
         "@nestjs/config": "^0.6.2",
         "@nestjs/core": "^7.5.1",
         "@nestjs/graphql": "^7.9.8",
+        "@nestjs/microservices": "^7.6.15",
         "@nestjs/platform-express": "^7.5.1",
         "@nestjs/typeorm": "^7.1.5",
         "@neuralegion/class-sanitizer": "^0.3.2",
@@ -33,6 +34,7 @@
         "observable-to-async-generator": "^1.0.1-rc",
         "pg": "^8.5.1",
         "ramda": "^0.27.1",
+        "redis": "^3.0.2",
         "reflect-metadata": "^0.1.13",
         "rimraf": "^3.0.2",
         "rxjs": "^6.6.3",
@@ -50,6 +52,7 @@
         "@types/jest": "^26.0.15",
         "@types/js-yaml": "^4.0.0",
         "@types/node": "^14.14.6",
+        "@types/redis": "^2.8.28",
         "@types/supertest": "^2.0.10",
         "@typescript-eslint/eslint-plugin": "^4.6.1",
         "@typescript-eslint/parser": "^4.6.1",
@@ -2721,6 +2724,64 @@
         "reflect-metadata": "^0.1.12"
       }
     },
+    "node_modules/@nestjs/microservices": {
+      "version": "7.6.15",
+      "resolved": "https://registry.npmjs.org/@nestjs/microservices/-/microservices-7.6.15.tgz",
+      "integrity": "sha512-WwjNvnudta+YKQOVnzkJGAggME1a8LsnoBZDzorYi4kCFs+U53viRsoidQUBR4oDYIu3IKkJmB9OCStHrK4PTw==",
+      "dependencies": {
+        "iterare": "1.2.1",
+        "json-socket": "0.3.0",
+        "tslib": "2.1.0"
+      },
+      "funding": {
+        "type": "opencollective",
+        "url": "https://opencollective.com/nest"
+      },
+      "peerDependencies": {
+        "@nestjs/common": "^7.0.0",
+        "@nestjs/core": "^7.0.0",
+        "@nestjs/websockets": "^7.0.0",
+        "amqp-connection-manager": "*",
+        "amqplib": "*",
+        "cache-manager": "*",
+        "grpc": "*",
+        "kafkajs": "*",
+        "mqtt": "*",
+        "nats": "*",
+        "redis": "*",
+        "reflect-metadata": "^0.1.12",
+        "rxjs": "^6.0.0"
+      },
+      "peerDependenciesMeta": {
+        "@nestjs/websockets": {
+          "optional": true
+        },
+        "amqp-connection-manager": {
+          "optional": true
+        },
+        "amqplib": {
+          "optional": true
+        },
+        "cache-manager": {
+          "optional": true
+        },
+        "grpc": {
+          "optional": true
+        },
+        "kafkajs": {
+          "optional": true
+        },
+        "mqtt": {
+          "optional": true
+        },
+        "nats": {
+          "optional": true
+        },
+        "redis": {
+          "optional": true
+        }
+      }
+    },
     "node_modules/@nestjs/platform-express": {
       "version": "7.6.15",
       "resolved": "https://registry.npmjs.org/@nestjs/platform-express/-/platform-express-7.6.15.tgz",
@@ -3428,6 +3489,15 @@
       "resolved": "https://registry.npmjs.org/@types/range-parser/-/range-parser-1.2.3.tgz",
       "integrity": "sha512-ewFXqrQHlFsgc09MK5jP5iR7vumV/BYayNC6PgJO2LPe8vrnNFyjQjSppfEngITi0qvfKtzFvgKymGheFM9UOA=="
     },
+    "node_modules/@types/redis": {
+      "version": "2.8.28",
+      "resolved": "https://registry.npmjs.org/@types/redis/-/redis-2.8.28.tgz",
+      "integrity": "sha512-8l2gr2OQ969ypa7hFOeKqtFoY70XkHxISV0pAwmQ2nm6CSPb1brmTmqJCGGrekCo+pAZyWlNXr+Kvo6L/1wijA==",
+      "dev": true,
+      "dependencies": {
+        "@types/node": "*"
+      }
+    },
     "node_modules/@types/serve-static": {
       "version": "1.13.9",
       "resolved": "https://registry.npmjs.org/@types/serve-static/-/serve-static-1.13.9.tgz",
@@ -10002,6 +10072,11 @@
       "integrity": "sha512-xbbCH5dCYU5T8LcEhhuh7HJ88HXuW3qsI3Y0zOZFKfZEHcpWiHU/Jxzk629Brsab/mMiHQti9wMP+845RPe3Vg==",
       "dev": true
     },
+    "node_modules/json-socket": {
+      "version": "0.3.0",
+      "resolved": "https://registry.npmjs.org/json-socket/-/json-socket-0.3.0.tgz",
+      "integrity": "sha512-jc8ZbUnYIWdxERFWQKVgwSLkGSe+kyzvmYxwNaRgx/c8NNyuHes4UHnPM3LUrAFXUx1BhNJ94n1h/KCRlbvV0g=="
+    },
     "node_modules/json-stable-stringify-without-jsonify": {
       "version": "1.0.1",
       "resolved": "https://registry.npmjs.org/json-stable-stringify-without-jsonify/-/json-stable-stringify-without-jsonify-1.0.1.tgz",
@@ -12035,6 +12110,24 @@
         "node": ">= 0.10"
       }
     },
+    "node_modules/redis": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/redis/-/redis-3.0.2.tgz",
+      "integrity": "sha512-PNhLCrjU6vKVuMOyFu7oSP296mwBkcE6lrAjruBYG5LgdSqtRBoVQIylrMyVZD/lkF24RSNNatzvYag6HRBHjQ==",
+      "dependencies": {
+        "denque": "^1.4.1",
+        "redis-commands": "^1.5.0",
+        "redis-errors": "^1.2.0",
+        "redis-parser": "^3.0.0"
+      },
+      "engines": {
+        "node": ">=6"
+      },
+      "funding": {
+        "type": "opencollective",
+        "url": "https://opencollective.com/node-redis"
+      }
+    },
     "node_modules/redis-commands": {
       "version": "1.7.0",
       "resolved": "https://registry.npmjs.org/redis-commands/-/redis-commands-1.7.0.tgz",
@@ -17792,6 +17885,16 @@
       "integrity": "sha512-TVtd/aTb7EqPhVczdeuvzF9dY0fyE3ivvCstc2eO+AkNqrfzSG1kXYYiUUznKjd0qDa8g2TmPSmHUQ21AXsV1Q==",
       "requires": {}
     },
+    "@nestjs/microservices": {
+      "version": "7.6.15",
+      "resolved": "https://registry.npmjs.org/@nestjs/microservices/-/microservices-7.6.15.tgz",
+      "integrity": "sha512-WwjNvnudta+YKQOVnzkJGAggME1a8LsnoBZDzorYi4kCFs+U53viRsoidQUBR4oDYIu3IKkJmB9OCStHrK4PTw==",
+      "requires": {
+        "iterare": "1.2.1",
+        "json-socket": "0.3.0",
+        "tslib": "2.1.0"
+      }
+    },
     "@nestjs/platform-express": {
       "version": "7.6.15",
       "resolved": "https://registry.npmjs.org/@nestjs/platform-express/-/platform-express-7.6.15.tgz",
@@ -18408,6 +18511,15 @@
       "resolved": "https://registry.npmjs.org/@types/range-parser/-/range-parser-1.2.3.tgz",
       "integrity": "sha512-ewFXqrQHlFsgc09MK5jP5iR7vumV/BYayNC6PgJO2LPe8vrnNFyjQjSppfEngITi0qvfKtzFvgKymGheFM9UOA=="
     },
+    "@types/redis": {
+      "version": "2.8.28",
+      "resolved": "https://registry.npmjs.org/@types/redis/-/redis-2.8.28.tgz",
+      "integrity": "sha512-8l2gr2OQ969ypa7hFOeKqtFoY70XkHxISV0pAwmQ2nm6CSPb1brmTmqJCGGrekCo+pAZyWlNXr+Kvo6L/1wijA==",
+      "dev": true,
+      "requires": {
+        "@types/node": "*"
+      }
+    },
     "@types/serve-static": {
       "version": "1.13.9",
       "resolved": "https://registry.npmjs.org/@types/serve-static/-/serve-static-1.13.9.tgz",
@@ -23465,6 +23577,11 @@
       "integrity": "sha512-xbbCH5dCYU5T8LcEhhuh7HJ88HXuW3qsI3Y0zOZFKfZEHcpWiHU/Jxzk629Brsab/mMiHQti9wMP+845RPe3Vg==",
       "dev": true
     },
+    "json-socket": {
+      "version": "0.3.0",
+      "resolved": "https://registry.npmjs.org/json-socket/-/json-socket-0.3.0.tgz",
+      "integrity": "sha512-jc8ZbUnYIWdxERFWQKVgwSLkGSe+kyzvmYxwNaRgx/c8NNyuHes4UHnPM3LUrAFXUx1BhNJ94n1h/KCRlbvV0g=="
+    },
     "json-stable-stringify-without-jsonify": {
       "version": "1.0.1",
       "resolved": "https://registry.npmjs.org/json-stable-stringify-without-jsonify/-/json-stable-stringify-without-jsonify-1.0.1.tgz",
@@ -25036,6 +25153,17 @@
         "resolve": "^1.1.6"
       }
     },
+    "redis": {
+      "version": "3.0.2",
+      "resolved": "https://registry.npmjs.org/redis/-/redis-3.0.2.tgz",
+      "integrity": "sha512-PNhLCrjU6vKVuMOyFu7oSP296mwBkcE6lrAjruBYG5LgdSqtRBoVQIylrMyVZD/lkF24RSNNatzvYag6HRBHjQ==",
+      "requires": {
+        "denque": "^1.4.1",
+        "redis-commands": "^1.5.0",
+        "redis-errors": "^1.2.0",
+        "redis-parser": "^3.0.0"
+      }
+    },
     "redis-commands": {
       "version": "1.7.0",
       "resolved": "https://registry.npmjs.org/redis-commands/-/redis-commands-1.7.0.tgz",

package.json

@@ -26,6 +26,7 @@
     "@nestjs/config": "^0.6.2",
     "@nestjs/core": "^7.5.1",
     "@nestjs/graphql": "^7.9.8",
+    "@nestjs/microservices": "^7.6.15",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/typeorm": "^7.1.5",
     "@neuralegion/class-sanitizer": "^0.3.2",
@@ -46,6 +47,7 @@
     "observable-to-async-generator": "^1.0.1-rc",
     "pg": "^8.5.1",
     "ramda": "^0.27.1",
+    "redis": "^3.0.2",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
     "rxjs": "^6.6.3",
@@ -63,6 +65,7 @@
     "@types/jest": "^26.0.15",
     "@types/js-yaml": "^4.0.0",
     "@types/node": "^14.14.6",
+    "@types/redis": "^2.8.28",
     "@types/supertest": "^2.0.10",
     "@typescript-eslint/eslint-plugin": "^4.6.1",
     "@typescript-eslint/parser": "^4.6.1",

src/commons/commons.constants.ts

@@ -0,0 +1 @@
+export const FENNEC_SERVICE = 'FENNEC_SERVICE';

src/main.ts

@@ -1,6 +1,7 @@
 import { ValidationPipe } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { NestFactory } from '@nestjs/core';
+import { Transport } from '@nestjs/microservices';
 import { AppModule } from './app.module';
 import { HttpExceptionFilter } from './commons/filters/all.exception-filter';
 import { SanitizePipe } from './commons/pipes/sanitize.pipe';
@@ -8,6 +9,19 @@ import { SanitizePipe } from './commons/pipes/sanitize.pipe';
 async function bootstrap() {
   const app = await NestFactory.create(AppModule, { bodyParser: false });
   const configService = app.get(ConfigService);
+
+  app.connectMicroservice({
+    transport: Transport.REDIS,
+    options: {
+      retryAttempts: 5,
+      retryDelay: 3000,
+      host: configService.get<string>('db.redis.host', 'localhost'),
+      port: configService.get<number>('db.redis.port', 6379),
+      password: configService.get<string>('db.redis.password', undefined),
+      prefix: configService.get<string>('db.redis.prefix', 'fennec') + ':',
+    },
+  });
+
   app.useGlobalPipes(new SanitizePipe());
   app.useGlobalPipes(
     new ValidationPipe({
@@ -15,6 +29,8 @@ async function bootstrap() {
     }),
   );
   app.useGlobalFilters(new HttpExceptionFilter());
+
+  await app.startAllMicroservicesAsync();
   await app.listen(configService.get<number>('http.port'));
 }
 bootstrap();

src/pipeline-tasks/pipeline-tasks.constants.ts

@@ -1,3 +1 @@
 export const PIPELINE_TASK_QUEUE = 'PIPELINE_TASK_QUEUE';
-export const PIPELINE_TASK_LOG_QUEUE = 'PIPELINE_TASK_LOG_QUEUE';
-export const PIPELINE_TASK_LOG_PUBSUB = 'PIPELINE_TASK_LOG_PUBSUB';

src/pipeline-tasks/pipeline-tasks.controller.spec.ts

@@ -0,0 +1,18 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { PipelineTasksController } from './pipeline-tasks.controller';
+
+describe('PipelineTasksController', () => {
+  let controller: PipelineTasksController;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [PipelineTasksController],
+    }).compile();
+
+    controller = module.get<PipelineTasksController>(PipelineTasksController);
+  });
+
+  it('should be defined', () => {
+    expect(controller).toBeDefined();
+  });
+});

src/pipeline-tasks/pipeline-tasks.controller.ts

@@ -0,0 +1,13 @@
+import { Controller } from '@nestjs/common';
+import { MessagePattern } from '@nestjs/microservices';
+import { interval, Observable } from 'rxjs';
+import { take } from 'rxjs/operators';
+
+@Controller('pipeline-tasks')
+export class PipelineTasksController {
+  @MessagePattern('subscriptionTaskCreated')
+  subscriptionTaskCreated(): Observable<number> {
+    console.log('subscriptionTaskCreated');
+    return interval(1000).pipe(take(10));
+  }
+}

src/pipeline-tasks/pipeline-tasks.module.ts

@@ -8,12 +8,12 @@ import { ReposModule } from '../repos/repos.module';
 import { RedisModule } from 'nestjs-redis';
 import { BullModule } from '@nestjs/bull';
 import { PipelineTaskConsumer } from './pipeline-task.consumer';
-import {
-  PIPELINE_TASK_QUEUE,
-  PIPELINE_TASK_LOG_PUBSUB,
-} from './pipeline-tasks.constants';
+import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
 import { PubSub } from 'apollo-server-express';
+import { PipelineTasksController } from './pipeline-tasks.controller';
+import { ClientsModule, Transport } from '@nestjs/microservices';
+import { FENNEC_SERVICE } from '../commons/commons.constants';
 
 @Module({
   imports: [
@@ -23,17 +23,17 @@ import { PubSub } from 'apollo-server-express';
     }),
     RedisModule,
     ReposModule,
+    ClientsModule.register([
+      { name: FENNEC_SERVICE, transport: Transport.REDIS },
+    ]),
   ],
   providers: [
     PipelineTasksService,
     PipelineTasksResolver,
     PipelineTaskConsumer,
     PipelineTaskLogsService,
-    {
-      provide: Symbol(PIPELINE_TASK_LOG_PUBSUB),
-      useValue: new PubSub(),
-    },
   ],
   exports: [PipelineTasksService],
+  controllers: [PipelineTasksController],
 })
 export class PipelineTasksModule {}

src/pipeline-tasks/pipeline-tasks.resolver.ts

@@ -1,16 +1,31 @@
-import { Resolver, Args, Mutation, Subscription, Query } from '@nestjs/graphql';
+import {
+  Resolver,
+  Args,
+  Mutation,
+  Subscription,
+  Query,
+  Int,
+} from '@nestjs/graphql';
 import { PipelineTask } from './pipeline-task.entity';
 import { PipelineTasksService } from './pipeline-tasks.service';
 import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
 import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
 import { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
 import { PipelineTaskLogsService } from './pipeline-task-logs.service';
+import { ClientProxy } from '@nestjs/microservices';
+import { Inject } from '@nestjs/common';
+import { FENNEC_SERVICE } from '../commons/commons.constants';
+import { observableToAsyncIterable } from '@graphql-tools/utils';
+import { mapTo, tap } from 'rxjs/operators';
+import { of } from 'rxjs';
 
 @Resolver()
 export class PipelineTasksResolver {
   constructor(
     private readonly service: PipelineTasksService,
     private readonly logsService: PipelineTaskLogsService,
+    @Inject(FENNEC_SERVICE)
+    private readonly client: ClientProxy,
   ) {}
 
   @Mutation(() => PipelineTask)
@@ -47,4 +62,16 @@ export class PipelineTasksResolver {
   async findPipelineTask(@Args('id') id: string) {
     return await this.service.findTaskById(id);
   }
+
+  @Subscription(() => Int, {
+    resolve: (value) => value,
+  })
+  async pipelineTaskCreated() {
+    return observableToAsyncIterable(
+      this.client.send('subscriptionTaskCreated', {}).pipe(
+        tap((val) => console.log(val)),
+        mapTo(1),
+      ),
+    );
+  }
 }