Merge pull request #3455 from RedisInsight/be/feature/RI-5803_abort_controller

#RI-5803 - add abort controller

Author: AmirAllayarovSofteq

redisinsight/api/src/modules/rdi/client/api.rdi.client.ts

@@ -1,6 +1,7 @@
 import axios, { AxiosInstance } from 'axios';
 import { plainToClass } from 'class-transformer';
 import { decode } from 'jsonwebtoken';
+import { Request } from 'express';
 
 import { RdiClient } from 'src/modules/rdi/client/rdi.client';
 import {
@@ -102,13 +103,21 @@ export class ApiRdiClient extends RdiClient {
     }
   }
 
-  async testConnections(config: string): Promise<RdiTestConnectionsResponseDto> {
+  async testConnections(config: string, req: Request): Promise<RdiTestConnectionsResponseDto> {
     try {
-      const response = await this.client.post(RdiUrl.TestConnections, config);
+      const abortController = new AbortController();
+      req.socket.on('close', () => {
+        abortController.abort();
+      });
+      const response = await this.client.post(
+        RdiUrl.TestConnections,
+        config,
+        { signal: abortController.signal },
+      );
 
       const actionId = response.data.action_id;
 
-      return this.pollActionStatus(actionId);
+      return this.pollActionStatus(actionId, abortController.signal);
     } catch (e) {
       throw wrapRdiPipelineError(e);
     }
@@ -169,15 +178,21 @@ export class ApiRdiClient extends RdiClient {
     }
   }
 
-  private async pollActionStatus(actionId: string): Promise<any> {
+  private async pollActionStatus(actionId: string, abortSignal: AbortSignal): Promise<any> {
     const startTime = Date.now();
     while (true) {
+      if (abortSignal.aborted) {
+        throw new RdiPipelineInternalServerErrorException();
+      }
       if (Date.now() - startTime > MAX_POLLING_TIME) {
         throw new RdiPipelineTimeoutException();
       }
 
       try {
-        const response = await this.client.get(`${RdiUrl.Action}/${actionId}`);
+        const response = await this.client.get(
+          `${RdiUrl.Action}/${actionId}`,
+          { signal: abortSignal },
+        );
         const { status, data, error } = response.data;
 
         if (status === 'failed') {

redisinsight/api/src/modules/rdi/client/rdi.client.ts

@@ -1,6 +1,7 @@
+import { Request } from 'express';
 import {
   Rdi,
-  RdiClientMetadata, RdiJob, RdiPipeline, RdiStatisticsResult,
+  RdiClientMetadata, RdiPipeline, RdiStatisticsResult,
 } from 'src/modules/rdi/models';
 import {
   RdiDryRunJobDto, RdiDryRunJobResponseDto, RdiTestConnectionsResponseDto,
@@ -36,7 +37,7 @@ export abstract class RdiClient {
 
   abstract dryRunJob(data: RdiDryRunJobDto): Promise<RdiDryRunJobResponseDto>;
 
-  abstract testConnections(config: string): Promise<RdiTestConnectionsResponseDto>;
+  abstract testConnections(config: string, req: Request): Promise<RdiTestConnectionsResponseDto>;
 
   abstract getStatistics(sections?: string): Promise<RdiStatisticsResult>;
 

redisinsight/api/src/modules/rdi/rdi-pipeline.controller.ts

@@ -1,10 +1,11 @@
 import {
   Body,
   ClassSerializerInterceptor, Controller, Get, Post, UseInterceptors, UsePipes, ValidationPipe,
-  Query,
+  Query, Req,
 } from '@nestjs/common';
 import { Rdi, RdiPipeline, RdiClientMetadata } from 'src/modules/rdi/models';
 import { ApiTags } from '@nestjs/swagger';
+import { Request } from 'express';
 import { ApiEndpoint } from 'src/decorators/api-endpoint.decorator';
 import { RdiPipelineService } from 'src/modules/rdi/rdi-pipeline.service';
 import { RequestRdiClientMetadata } from 'src/modules/rdi/decorators';
@@ -72,10 +73,11 @@ export class RdiPipelineController {
     responses: [{ status: 200, type: RdiTestConnectionsResponseDto }],
   })
   async testConnections(
-    @RequestRdiClientMetadata() rdiClientMetadata: RdiClientMetadata,
+    @Req() req: Request,
+      @RequestRdiClientMetadata() rdiClientMetadata: RdiClientMetadata,
       @Body() config: string,
   ): Promise<RdiTestConnectionsResponseDto> {
-    return this.rdiPipelineService.testConnections(rdiClientMetadata, config);
+    return this.rdiPipelineService.testConnections(rdiClientMetadata, config, req);
   }
 
   @Get('/strategies')

redisinsight/api/src/modules/rdi/rdi-pipeline.service.ts

@@ -1,4 +1,5 @@
 import { Injectable, Logger } from '@nestjs/common';
+import { Request } from 'express';
 import { RdiClientMetadata, RdiPipeline } from 'src/modules/rdi/models';
 import { RdiClientProvider } from 'src/modules/rdi/providers/rdi.client.provider';
 import { RdiDryRunJobDto, RdiTestConnectionsResponseDto } from 'src/modules/rdi/dto';
@@ -69,12 +70,12 @@ export class RdiPipelineService {
     }
   }
 
-  async testConnections(rdiClientMetadata: RdiClientMetadata, config: string): Promise<RdiTestConnectionsResponseDto> {
+  async testConnections(rdiClientMetadata: RdiClientMetadata, config: string, req: Request): Promise<RdiTestConnectionsResponseDto> {
     this.logger.log('Trying to test connections');
 
     const client = await this.rdiClientProvider.getOrCreate(rdiClientMetadata);
 
-    return await client.testConnections(config);
+    return await client.testConnections(config, req);
   }
 
   async getStrategies(rdiClientMetadata: RdiClientMetadata): Promise<object> {