implemented pipeline module on gateway ( still have to implement API documentation properly)

src/app.module.ts

@@ -4,11 +4,12 @@ import { UsersGrpcClientModule } from './users-grpc-client/users-grpc-client.mod
 import { OutputsModule } from './outputs/outputs.module';
 import { EnrichmentModule } from './enrichment/enrichment.module';
 import { TransformationsModule } from './transformations/transformations.module';
+import { PipelinesModule } from './pipelines/pipelines.module';
 
 @Module({
   imports: [ 
     InputGrpcClientModule, 
-    OutputsModule, EnrichmentModule, TransformationsModule
+    OutputsModule, EnrichmentModule, TransformationsModule, PipelinesModule
   ],
   controllers: [],
   providers: [],

src/pipelines/config/grpc-client.ts

@@ -0,0 +1,17 @@
+import { join } from 'path';
+import { ClientOptions, Transport } from '@nestjs/microservices';
+
+export const GrpcClientConfiguration: ClientOptions = {
+  transport: Transport.GRPC,
+  options: {
+    //url: `${process.env.USERS_SVC_URL}:${process.env.USERS_SVC_PORT}`,
+    url: `127.0.0.1:50054`,
+    package: 'pipeline',
+    protoPath: join(__dirname, '..', 'proto', 'pipeline.proto'),
+    loader: {
+      enums: String,
+      objects: true,
+      arrays: true,
+    },
+  },
+};

src/pipelines/dto/create-pipeline.dto.ts

@@ -0,0 +1,12 @@
+import { InputGrpcClient } from "src/inputs/entities/input-grcp-client.entity";
+import { Output } from "src/outputs/entities/output.entity";
+import { Transformation } from "src/transformations/entities/transformation.entity";
+
+export class CreatePipelineDto {
+    input:InputGrpcClient;
+    output:Output;
+    transformations:Transformation;
+    tags:string[];
+    name:string;
+    description:string;
+}

src/pipelines/dto/update-pipeline.dto.ts

@@ -0,0 +1,6 @@
+import { PartialType } from '@nestjs/mapped-types';
+import { CreatePipelineDto } from './create-pipeline.dto';
+
+export class UpdatePipelineDto extends PartialType(CreatePipelineDto) {
+
+}

src/pipelines/entities/pipeline.entity.ts

@@ -0,0 +1,14 @@
+import { InputGrpcClient } from "src/inputs/entities/input-grcp-client.entity";
+import { Output } from "src/outputs/entities/output.entity";
+import { Transformation } from "src/transformations/entities/transformation.entity";
+
+export class Pipeline {
+    id:string;
+    input:InputGrpcClient;
+    output:Output;
+    transformations:Transformation;
+    tags:string[];
+    name:string;
+    description:string;
+    status:string;
+}

src/pipelines/pipelines.controller.spec.ts

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

src/pipelines/pipelines.controller.ts

@@ -0,0 +1,56 @@
+import { Controller, Delete, Get, OnModuleInit, Param, Post, Put } from '@nestjs/common';
+import { Client, ClientGrpc, MessagePattern, Payload } from '@nestjs/microservices';
+import { PipelineGrpcClientService } from './pipelines.service';
+import { CreatePipelineDto } from './dto/create-pipeline.dto';
+import { UpdatePipelineDto } from './dto/update-pipeline.dto';
+import { GrpcClientConfiguration } from './config/grpc-client';
+import { ApiTags } from '@nestjs/swagger';
+
+@ApiTags('pipelines')
+@Controller('pipeline')
+export class PipelinesController implements OnModuleInit{
+  
+  @Client(GrpcClientConfiguration)
+  private pipelineServiceClient: ClientGrpc;
+
+  private pipelineService: PipelineGrpcClientService;
+
+  onModuleInit() {
+    this.pipelineService =
+      this.pipelineServiceClient.getService<PipelineGrpcClientService>('PipelineService');
+  }
+
+  @Post()
+  create(@Payload() createPipelineDto: CreatePipelineDto) {
+    return this.pipelineService.create(createPipelineDto);
+  }
+
+  @Get()
+  list() {
+    return this.pipelineService.list();
+  }
+
+  @Get(':id')
+  show(@Param() id: string) {
+    return this.pipelineService.show(id);
+  }
+
+  @Put(':id')
+  update(@Param() id:string,@Payload() updatePipelineDto: UpdatePipelineDto) {
+    return this.pipelineService.update(id, updatePipelineDto);
+  }
+
+  @Delete()
+  remove(@Param() id:string) {
+    return this.pipelineService.remove(id);
+  }
+
+  @Post()
+  getPipelineLogsMessages(@Param() id:string,@Payload() details:string){
+    return this.pipelineService.getPipelineLogsMessages(id,details);
+  }
+
+  getPipelineStatus(id:string){
+    return this.pipelineService.getPipelineStatus(id)
+  }
+}

src/pipelines/pipelines.module.ts

@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { PipelineGrpcClientService } from './pipelines.service';
+import { PipelinesController } from './pipelines.controller';
+
+@Module({
+  controllers: [PipelinesController],
+  providers: [PipelineGrpcClientService]
+})
+export class PipelinesModule {}

src/pipelines/pipelines.service.spec.ts

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

src/pipelines/pipelines.service.ts

@@ -0,0 +1,34 @@
+import { Injectable } from '@nestjs/common';
+import { CreatePipelineDto } from './dto/create-pipeline.dto';
+import { UpdatePipelineDto } from './dto/update-pipeline.dto';
+
+@Injectable()
+export class PipelineGrpcClientService {
+  create(createPipelineDto: CreatePipelineDto) {
+    return 'This action adds a new pipeline';
+  }
+
+  list() {
+    return `This action returns all pipelines`;
+  }
+
+  show(id: string) {
+    return `This action returns a #${id} pipeline`;
+  }
+
+  update(id: string, updatePipelineDto: UpdatePipelineDto) {
+    return `This action updates a #${id} pipeline`;
+  }
+
+  remove(id: string) {
+    return `This action removes a #${id} pipeline`;
+  }
+
+  getPipelineLogsMessages(id:string,details:string){
+    return `This action perfoms a getLogMessages on #${id} pipeline`;
+  }
+
+  getPipelineStatus(id:string){
+    return `This action performs a getStatus on #${id} pipeline`;
+  }
+}

src/pipelines/proto/pipeline.proto

@@ -0,0 +1,97 @@
+syntax = "proto3";
+
+package pipeline;
+
+import "google/protobuf/timestamp.proto";
+
+service PipelineService{
+    rpc Create(CreatePipelineRequest) returns (Pipeline) {}
+    rpc Show(IdRequest) returns (Pipeline) {}
+    rpc List(Empty) returns (ListResponse){}
+    rpc Update(UpdatePipelineRequest) returns (Pipeline){}
+    rpc Remove(IdRequest) returns (Empty){}
+    rpc getPipelineLogsMessages(getPipelineLogsMessagesRequest) returns (getPipelineLogsMessagesResponse){}
+    rpc getPipelineStatus(IdRequest) returns (getPipelineStatusResponse){}
+}
+
+
+message Empty {}
+
+message IdRequest{
+  string id = 1;
+}
+
+message Transformation {
+    string id = 1;
+    string inputSource = 2;
+    string table = 3;
+    string type = 4;
+    string params = 5;
+}
+
+message Input {
+    string id = 1;
+    string cronFormated = 2;
+    string name = 3;
+    string plugin = 4;
+    string values = 5;
+    string operation = 6;
+}
+  
+message Output {
+    string id = 1;
+    string name = 2;
+    string plugin = 3;
+    string values = 4;
+    string operation = 5;
+}
+
+message Pipeline {
+  string id = 1;
+  Input input = 2;
+  Output output = 3;
+  Transformation transformations = 4;
+  repeated string tags = 5;
+  string name = 6;
+  string description = 7;
+  string status = 8;
+}
+
+message CreatePipelineRequest {
+    Input input = 1;
+    Output output = 2;
+    Transformation transformations = 3;
+    repeated string tags = 4;
+    string name = 5;
+    string description = 6;
+}
+
+message UpdatePipelineRequest{
+    string id = 1;
+    repeated string tags = 2;
+    string name = 3;
+    string description = 4;
+    string status = 5;
+}
+
+message ListResponse {
+  repeated Pipeline pipelines = 1;
+}
+
+message getPipelineLogsMessagesRequest{
+    string id = 1;
+    string details = 2; 
+}
+
+message getPipelineLogsMessagesResponseItem{
+    string date = 1;
+    string log = 2; 
+}
+
+message getPipelineLogsMessagesResponse{
+    repeated getPipelineLogsMessagesResponseItem pipelineLogs = 1;
+}
+
+message getPipelineStatusResponse{
+    string status = 1;
+}