Merge pull request #745 from RedisInsight/feature/RI-1453-pub-sub

#RI-3030 BE for subscribe/unsubscribe base implementation

Author: arthosofteq

redisinsight/api/src/app.module.ts

@@ -12,6 +12,7 @@ import { PluginModule } from 'src/modules/plugin/plugin.module';
 import { CommandsModule } from 'src/modules/commands/commands.module';
 import { WorkbenchModule } from 'src/modules/workbench/workbench.module';
 import { SlowLogModule } from 'src/modules/slow-log/slow-log.module';
+import { PubSubModule } from 'src/modules/pub-sub/pub-sub.module';
 import { SharedModule } from './modules/shared/shared.module';
 import { InstancesModule } from './modules/instances/instances.module';
 import { BrowserModule } from './modules/browser/browser.module';
@@ -43,6 +44,7 @@ const PATH_CONFIG = config.get('dir_path');
     PluginModule,
     CommandsModule,
     ProfilerModule,
+    PubSubModule,
     SlowLogModule,
     EventEmitterModule.forRoot(),
     ...(SERVER_CONFIG.staticContent

redisinsight/api/src/app.routes.ts

@@ -6,6 +6,7 @@ import { RedisSentinelModule } from 'src/modules/redis-sentinel/redis-sentinel.m
 import { CliModule } from 'src/modules/cli/cli.module';
 import { WorkbenchModule } from 'src/modules/workbench/workbench.module';
 import { SlowLogModule } from 'src/modules/slow-log/slow-log.module';
+import { PubSubModule } from 'src/modules/pub-sub/pub-sub.module';
 
 export const routes: Routes = [
   {
@@ -28,6 +29,10 @@ export const routes: Routes = [
         path: '/:dbInstance',
         module: SlowLogModule,
       },
+      {
+        path: '/:dbInstance',
+        module: PubSubModule,
+      },
     ],
   },
   {

redisinsight/api/src/modules/cli/utils/getUnsupportedCommands.spec.ts

@@ -2,7 +2,7 @@ import { getUnsupportedCommands } from './getUnsupportedCommands';
 
 describe('cli unsupported commands', () => {
   it('should return correct list', () => {
-    const expectedResult = ['monitor', 'subscribe', 'psubscribe', 'sync', 'psync', 'script debug'];
+    const expectedResult = ['monitor', 'subscribe', 'psubscribe', 'ssubscribe', 'sync', 'psync', 'script debug'];
 
     expect(getUnsupportedCommands()).toEqual(expectedResult);
   });

redisinsight/api/src/modules/cli/utils/getUnsupportedCommands.ts

@@ -6,6 +6,7 @@ export enum CliToolUnsupportedCommands {
   Monitor = 'monitor',
   Subscribe = 'subscribe',
   PSubscribe = 'psubscribe',
+  SSubscribe = 'ssubscribe',
   Sync = 'sync',
   PSync = 'psync',
   ScriptDebug = 'script debug',

redisinsight/api/src/modules/profiler/models/log-file.ts

@@ -124,7 +124,7 @@ export class LogFile {
       this.writeStream?.close();
       this.writeStream = null;
       const size = this.getFileSize();
-      fs.unlink(this.filePath);
+      fs.unlinkSync(this.filePath);
 
       this.analyticsEvents.get(TelemetryEvents.ProfilerLogDeleted)(this.instanceId, size);
     } catch (e) {

redisinsight/api/src/modules/pub-sub/constants/index.ts

@@ -0,0 +1,28 @@
+export enum PubSubClientEvents {
+  Subscribe = 'subscribe',
+  Unsubscribe = 'unsubscribe',
+}
+
+export enum PubSubServerEvents {
+  Exception = 'exception',
+}
+
+export enum SubscriptionType {
+  Subscribe = 's',
+  PSubscribe = 'p',
+  SSubscribe = 'ss',
+}
+
+export enum RedisClientStatus {
+  Connecting = 'connecting',
+  Connected = 'connected',
+  Error = 'error',
+  End = 'end',
+}
+
+export enum RedisClientEvents {
+  Connected = 'connected',
+  ConnectionError = 'connection_error',
+  Message = 'message',
+  End = 'end',
+}

redisinsight/api/src/modules/pub-sub/decorators/client.decorator.ts

@@ -0,0 +1,11 @@
+import { get } from 'lodash';
+import { createParamDecorator, ExecutionContext } from '@nestjs/common';
+import { UserClient } from 'src/modules/pub-sub/model/user-client';
+
+export const Client = createParamDecorator(
+  (data: unknown, ctx: ExecutionContext): UserClient => {
+    const socket = ctx.switchToWs().getClient();
+
+    return new UserClient(socket.id, socket, get(socket, 'handshake.query.instanceId'));
+  },
+);

redisinsight/api/src/modules/pub-sub/dto/index.ts

@@ -0,0 +1,3 @@
+export * from './subscribe.dto';
+export * from './subscription.dto';
+export * from './messages.response';

redisinsight/api/src/modules/pub-sub/dto/messages.response.ts

@@ -0,0 +1,7 @@
+import { IMessage } from 'src/modules/pub-sub/interfaces/message.interface';
+
+export class MessagesResponse {
+  messages: IMessage[];
+
+  count: number;
+}

redisinsight/api/src/modules/pub-sub/dto/publish.dto.ts

@@ -0,0 +1,24 @@
+import {
+  IsNotEmpty, IsString,
+} from 'class-validator';
+import { ApiProperty } from '@nestjs/swagger';
+
+export class PublishDto {
+  @ApiProperty({
+    type: String,
+    description: 'Message to send',
+    example: '{"hello":"world"}',
+  })
+  @IsNotEmpty()
+  @IsString()
+  message: string;
+
+  @ApiProperty({
+    type: String,
+    description: 'Chanel name',
+    example: 'channel-1',
+  })
+  @IsNotEmpty()
+  @IsString()
+  channel: string;
+}