#RI-2935 BE Show Consumers List

Author: Artem

redisinsight/api/src/modules/browser/browser.module.ts

@@ -6,6 +6,8 @@ import { StreamController } from 'src/modules/browser/controllers/stream/stream.
 import { StreamService } from 'src/modules/browser/services/stream/stream.service';
 import { ConsumerGroupController } from 'src/modules/browser/controllers/stream/consumer-group.controller';
 import { ConsumerGroupService } from 'src/modules/browser/services/stream/consumer-group.service';
+import { ConsumerController } from 'src/modules/browser/controllers/stream/consumer.controller';
+import { ConsumerService } from 'src/modules/browser/services/stream/consumer.service';
 import { HashController } from './controllers/hash/hash.controller';
 import { KeysController } from './controllers/keys/keys.controller';
 import { KeysBusinessService } from './services/keys-business/keys-business.service';
@@ -35,6 +37,7 @@ import { BrowserToolClusterService } from './services/browser-tool-cluster/brows
     HashController,
     StreamController,
     ConsumerGroupController,
+    ConsumerController,
   ],
   providers: [
     KeysBusinessService,
@@ -46,6 +49,7 @@ import { BrowserToolClusterService } from './services/browser-tool-cluster/brows
     HashBusinessService,
     StreamService,
     ConsumerGroupService,
+    ConsumerService,
     BrowserToolService,
     BrowserToolClusterService,
   ],
@@ -62,6 +66,9 @@ export class BrowserModule implements NestModule {
         RouterModule.resolvePath(SetController),
         RouterModule.resolvePath(ZSetController),
         RouterModule.resolvePath(RejsonRlController),
+        RouterModule.resolvePath(StreamController),
+        RouterModule.resolvePath(ConsumerGroupController),
+        RouterModule.resolvePath(ConsumerController),
       );
   }
 }

redisinsight/api/src/modules/browser/constants/browser-tool-commands.ts

@@ -81,6 +81,7 @@ export enum BrowserToolStreamCommands {
   XAdd = 'xadd',
   XDel = 'xdel',
   XInfoGroups = 'xinfo groups',
+  XInfoConsumers = 'xinfo consumers',
   XPending = 'xpending',
   XGroupCreate = 'xgroup create',
   XGroupSetId = 'xgroup setid',

redisinsight/api/src/modules/browser/controllers/stream/consumer-group.controller.ts

@@ -26,7 +26,7 @@ export class ConsumerGroupController {
 
   @Post('/get')
   @ApiRedisInstanceOperation({
-    description: 'Get stream entries',
+    description: 'Get consumer groups list',
     statusCode: 200,
     responses: [
       {

redisinsight/api/src/modules/browser/controllers/stream/consumer.controller.ts

@@ -0,0 +1,43 @@
+import {
+  Body,
+  Controller,
+  Param,
+  Post,
+  UsePipes,
+  ValidationPipe,
+} from '@nestjs/common';
+import { ApiTags } from '@nestjs/swagger';
+import { ApiRedisInstanceOperation } from 'src/decorators/api-redis-instance-operation.decorator';
+import {
+  ConsumerDto,
+  ConsumerGroupDto,
+  GetConsumersDto,
+} from 'src/modules/browser/dto/stream.dto';
+import { ConsumerService } from 'src/modules/browser/services/stream/consumer.service';
+
+@ApiTags('Streams')
+@Controller('streams/consumer-groups/consumers')
+@UsePipes(new ValidationPipe({ transform: true }))
+export class ConsumerController {
+  constructor(private service: ConsumerService) {}
+
+  @Post('/get')
+  @ApiRedisInstanceOperation({
+    description: 'Get stream entries',
+    statusCode: 200,
+    responses: [
+      {
+        status: 200,
+        description: 'Returns stream consumer groups.',
+        type: ConsumerGroupDto,
+        isArray: true,
+      },
+    ],
+  })
+  async getConsumers(
+    @Param('dbInstance') instanceId: string,
+      @Body() dto: GetConsumersDto,
+  ): Promise<ConsumerDto[]> {
+    return this.service.getConsumers({ instanceId }, dto);
+  }
+}

redisinsight/api/src/modules/browser/dto/stream.dto.ts

@@ -289,3 +289,38 @@ export class DeleteConsumerGroupsResponse {
   })
   affected: number;
 }
+
+export class ConsumerDto {
+  @ApiProperty({
+    type: String,
+    description: 'The consumer\'s name',
+    example: 'consumer-1',
+  })
+  name: string;
+
+  @ApiProperty({
+    type: Number,
+    description: 'The number of pending messages for the client, '
+      + 'which are messages that were delivered but are yet to be acknowledged',
+    example: 2,
+  })
+  pending: number = 0;
+
+  @ApiProperty({
+    type: Number,
+    description: 'The number of milliseconds that have passed since the consumer last interacted with the server',
+    example: 22442,
+  })
+  idle: number = 0;
+}
+
+export class GetConsumersDto extends KeyDto {
+  @ApiProperty({
+    type: String,
+    description: 'Consumer group name',
+    example: 'group-1',
+  })
+  @IsNotEmpty()
+  @IsString()
+  groupName: string;
+}

redisinsight/api/src/modules/browser/services/stream/consumer.service.ts

@@ -0,0 +1,109 @@
+import {
+  BadRequestException, Injectable, Logger, NotFoundException,
+} from '@nestjs/common';
+import { IFindRedisClientInstanceByOptions } from 'src/modules/core/services/redis/redis.service';
+import { RedisErrorCodes } from 'src/constants';
+import { catchAclError, convertStringsArrayToObject } from 'src/utils';
+import {
+  BrowserToolKeysCommands, BrowserToolStreamCommands,
+} from 'src/modules/browser/constants/browser-tool-commands';
+import { BrowserToolService } from 'src/modules/browser/services/browser-tool/browser-tool.service';
+import ERROR_MESSAGES from 'src/constants/error-messages';
+import {
+  ConsumerDto,
+  GetConsumersDto,
+} from 'src/modules/browser/dto/stream.dto';
+
+@Injectable()
+export class ConsumerService {
+  private logger = new Logger('ConsumerService');
+
+  constructor(private browserTool: BrowserToolService) {}
+
+  /**
+   * Get consumers list inside particular group
+   * @param clientOptions
+   * @param dto
+   */
+  async getConsumers(
+    clientOptions: IFindRedisClientInstanceByOptions,
+    dto: GetConsumersDto,
+  ): Promise<ConsumerDto[]> {
+    try {
+      this.logger.log('Getting consumers list.');
+
+      const exists = await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolKeysCommands.Exists,
+        [dto.keyName],
+      );
+
+      if (!exists) {
+        return Promise.reject(new NotFoundException(ERROR_MESSAGES.KEY_NOT_EXIST));
+      }
+
+      return ConsumerService.formatReplyToDto(await this.browserTool.execCommand(
+        clientOptions,
+        BrowserToolStreamCommands.XInfoConsumers,
+        [dto.keyName, dto.groupName],
+      ));
+    } catch (error) {
+      if (error instanceof NotFoundException) {
+        throw error;
+      }
+
+      if (error?.message.includes(RedisErrorCodes.WrongType)) {
+        throw new BadRequestException(error.message);
+      }
+
+      throw catchAclError(error);
+    }
+  }
+
+  /**
+   * Converts RESP response from Redis
+   * [
+   *  ['name', 'consumer-1', 'pending', 0, 'idle', 258741],
+   *  ['name', 'consumer-2', 'pending', 0, 'idle', 258741],
+   *  ...
+   * ]
+   *
+   * to DTO
+   *
+   * [
+   *  {
+   *    name: 'consumer-1',
+   *    pending: 0,
+   *    idle: 258741,
+   *  },
+   *  {
+   *    name: 'consumer-2',
+   *    pending: 0,
+   *    idle: 258741,
+   *  },
+   *   ...
+   * ]
+   * @param reply
+   */
+  static formatReplyToDto(reply: Array<Array<string | number>>): ConsumerDto[] {
+    return reply.map(ConsumerService.formatArrayToDto);
+  }
+
+  /**
+   * Format single reply entry to DTO
+   * @param entry
+   */
+  static formatArrayToDto(entry: Array<string | number>): ConsumerDto {
+    if (!entry?.length) {
+      return null;
+    }
+
+    const entryObj = convertStringsArrayToObject(entry as string[]);
+
+    return {
+      name: entryObj['name'],
+      pending: entryObj['pending'],
+      idle: entryObj['idle'],
+    };
+  }
+}