add more logs

Author: Artem

redisinsight/api/src/modules/pub-sub/model/abstract.subscription.ts

@@ -6,6 +6,10 @@ import { ISubscription } from 'src/modules/pub-sub/interfaces/subscription.inter
 import { IMessage } from 'src/modules/pub-sub/interfaces/message.interface';
 import * as IORedis from 'ioredis';
 
+const EMIT_WAIT = 30;
+const EMIT_MAX_WAIT = 100;
+const MESSAGES_MAX = 5000;
+
 export abstract class AbstractSubscription implements ISubscription {
   protected readonly id: string;
 
@@ -26,34 +30,44 @@ export abstract class AbstractSubscription implements ISubscription {
     this.id = `${this.type}:${this.channel}`;
     this.debounce = debounce(() => {
       if (this.messages.length) {
-        this.userClient.getSocket().emit(this.id, {
-          messages: this.messages.slice(0, 5000),
-          count: this.messages.length,
-        } as MessagesResponse);
+        this.userClient.getSocket()
+          .emit(this.id, {
+            messages: this.messages.slice(0, MESSAGES_MAX),
+            count: this.messages.length,
+          } as MessagesResponse);
         this.messages = [];
       }
-    }, 30, {
-      maxWait: 100,
+    }, EMIT_WAIT, {
+      maxWait: EMIT_MAX_WAIT,
     });
   }
 
-  getId() { return this.id; }
-
-  getChannel() { return this.channel; }
-
-  getType() { return this.type; }
+  getId() {
+    return this.id;
+  }
 
-  async subscribe(client: IORedis.Redis | IORedis.Cluster): Promise<void> {
-    throw new Error('"subscribe" method should be implemented');
+  getChannel() {
+    return this.channel;
   }
 
-  async unsubscribe(client: IORedis.Redis | IORedis.Cluster): Promise<void> {
-    throw new Error('"unsubscribe" method should be implemented');
+  getType() {
+    return this.type;
   }
 
+  abstract subscribe(client: IORedis.Redis | IORedis.Cluster): Promise<void>;
+
+  abstract unsubscribe(client: IORedis.Redis | IORedis.Cluster): Promise<void>;
+
   pushMessage(message: IMessage) {
     this.messages.push(message);
 
     this.debounce();
   }
+
+  toString() {
+    return `${this.constructor.name}:${JSON.stringify({
+      id: this.id,
+      mL: this.messages.length,
+    })}`;
+  }
 }

redisinsight/api/src/modules/pub-sub/model/user-session.ts

@@ -38,12 +38,15 @@ export class UserSession {
    * @param subscription
    */
   async subscribe(subscription: ISubscription) {
+    this.logger.debug(`Subscribe ${subscription} ${this}. Getting Redis client...`);
+
     const client = await this.redisClient?.getClient();
 
     if (!client) { throw new Error('There is no Redis client initialized'); }
 
     if (!this.subscriptions.has(subscription.getId())) {
       this.subscriptions.set(subscription.getId(), subscription);
+      this.logger.debug(`Subscribe to Redis ${subscription} ${this}`);
       await subscription.subscribe(client);
     }
   }
@@ -54,14 +57,18 @@ export class UserSession {
    * @param subscription
    */
   async unsubscribe(subscription: ISubscription) {
+    this.logger.debug(`Unsubscribe ${subscription} ${this}`);
+
     this.subscriptions.delete(subscription.getId());
 
     const client = await this.redisClient?.getClient();
 
     if (client) {
+      this.logger.debug(`Unsubscribe from Redis ${subscription} ${this}`);
       await subscription.unsubscribe(client);
 
       if (!this.subscriptions.size) {
+        this.logger.debug(`Unsubscribe: Destroy RedisClient ${this}`);
         this.redisClient.destroy();
       }
     }
@@ -88,6 +95,8 @@ export class UserSession {
    * to be sure that there is no open connections left
    */
   handleDisconnect() {
+    this.logger.debug(`Handle disconnect ${this}`);
+
     this.userClient.getSocket().emit(
       PubSubServerEvents.Exception,
       new PubSubWsException(ERROR_MESSAGES.NO_CONNECTION_TO_REDIS_DB),
@@ -104,6 +113,8 @@ export class UserSession {
 
     this.subscriptions = new Map();
     this.redisClient.destroy();
+
+    this.logger.debug(`Destroyed ${this}`);
   }
 
   toString() {

redisinsight/api/src/modules/pub-sub/pub-sub.service.ts

@@ -27,6 +27,8 @@ export class PubSubService {
    */
   async subscribe(userClient: UserClient, dto: SubscribeDto) {
     try {
+      this.logger.log('Subscribing to channels(s)');
+
       const session = await this.sessionProvider.getOrCreateUserSession(userClient);
       await Promise.all(dto.subscriptions.map((subDto) => session.subscribe(
         this.subscriptionProvider.createSubscription(userClient, subDto),
@@ -49,6 +51,8 @@ export class PubSubService {
    */
   async unsubscribe(userClient: UserClient, dto: SubscribeDto) {
     try {
+      this.logger.log('Unsubscribing from channels(s)');
+
       const session = await this.sessionProvider.getOrCreateUserSession(userClient);
       await Promise.all(dto.subscriptions.map((subDto) => session.unsubscribe(
         this.subscriptionProvider.createSubscription(userClient, subDto),
@@ -120,6 +124,7 @@ export class PubSubService {
    * @param id
    */
   async handleDisconnect(id: string) {
+    this.logger.log(`Handle disconnect event: ${id}`);
     const session = this.sessionProvider.getUserSession(id);
 
     if (session) {

redisinsight/api/test/api/ws/pub-sub/pub-sub.test.ts

@@ -100,7 +100,6 @@ describe('pub-sub', function () {
           expect(ack).to.eql({ status: 'ok' });
 
           client.on('s:channel-a', (data) => {
-            console.log('___data', data);
             expect(data.count).to.be.eql(1);
             expect(data.messages.length).to.be.eql(1);
             const [message] = data.messages;