refactor code

Author: Ihar

common/src/mq/nats-service.ts

@@ -1,7 +1,7 @@
-import { NatsConnection, headers, Subscription } from 'nats';
-import { GenerateUUIDv4 } from '@guardian/interfaces';
-import { ZipCodec } from './zip-codec.js';
-import { IMessageResponse } from '../models/index.js';
+import {NatsConnection, headers, Subscription} from 'nats';
+import {GenerateUUIDv4} from '@guardian/interfaces';
+import {ZipCodec} from './zip-codec.js';
+import {IMessageResponse} from '../models/index.js';
 
 type CallbackFunction = (body: any, error?: string, code?: number) => void;
 
@@ -56,28 +56,25 @@ export abstract class NatsService {
             throw new Error('Connection must set first');
         }
         this.connection.subscribe(this.replySubject, {
-            callback: this.callbackSubscribe
-        });
-    }
-
-    public callbackSubscribe = async (error, msg) => {
-        if (!error) {
-            const messageId = msg.headers.get('messageId');
-            const fn = this.responseCallbacksMap.get(messageId);
-            if (fn) {
-                const message = (await this.codec.decode(msg.data)) as IMessageResponse<any>;
-                if (!message) {
-                    fn(null);
+            callback: async (error, msg) => {
+                if (!error) {
+                    const messageId = msg.headers.get('messageId');
+                    const fn = this.responseCallbacksMap.get(messageId);
+                    if (fn) {
+                        const message = (await this.codec.decode(msg.data)) as IMessageResponse<any>;
+                        if (!message) {
+                            fn(null)
+                        } else {
+                            fn(message.body, message.error, message.code);
+                        }
+                        this.responseCallbacksMap.delete(messageId)
+                    }
                 } else {
-                    fn(message.body, message.error, message.code);
+                    console.error(error);
                 }
-                this.responseCallbacksMap.delete(messageId);
             }
-        } else {
-            console.error(error);
-        }
-    };
-
+        });
+    }
 
     /**
      * Set connection
@@ -129,19 +126,25 @@ export abstract class NatsService {
      * Send message
      * @param subject
      * @param data
+     * @param isResponseCallback
      */
-    public sendMessage<T>(subject: string, data?: unknown): Promise<T> {
+    public sendMessage<T>(subject: string, data?: unknown, isResponseCallback: boolean = true): Promise<T> {
+        console.log('isResponseCallback', isResponseCallback)
         const messageId = GenerateUUIDv4();
         return new Promise(async (resolve, reject) => {
             const head = headers();
             head.append('messageId', messageId);
-            this.responseCallbacksMap.set(messageId, (body: T, error?: string, code?: number) => {
-                if (error) {
-                    reject(new MessageError(error, code));
-                } else {
-                    resolve(body);
-                }
-            })
+            if (isResponseCallback) {
+                this.responseCallbacksMap.set(messageId, (body: T, error?: string, code?: number) => {
+                    if (error) {
+                        reject(new MessageError(error, code));
+                    } else {
+                        resolve(body);
+                    }
+                })
+            } else {
+                resolve(null);
+            }
 
             this.connection.publish(subject, await this.codec.encode(data), {
                 reply: this.replySubject,
@@ -194,17 +197,6 @@ export abstract class NatsService {
         });
     }
 
-    getMapSizeInMB(map: Map<any, any>): number {
-        try {
-            const jsonString = JSON.stringify([...map]);
-            const bytes = new TextEncoder().encode(jsonString).length;
-            return bytes / (1024 * 1024);
-        } catch (error) {
-            console.error("Error calculating Map size:", error);
-            return 0;
-        }
-    }
-
     /**
      * Get messages
      * @param subject
@@ -229,8 +221,6 @@ export abstract class NatsService {
                         cb(await this.codec.decode(msg.data), msg.headers);
                     }
                     console.log(`📊 Active subscriptions: ${this.responseCallbacksMap.size}`);
-
-                    console.log(`💾 Map size: ${this.getMapSizeInMB(this.responseCallbacksMap).toFixed(4)} MB`);
                 } catch (error) {
                     console.error(error);
                 }

worker-service/src/api/worker.ts

@@ -152,11 +152,6 @@ export class Worker extends NatsService {
     public async init(): Promise<void> {
         await super.init();
         this.channel = new MessageBrokerChannel(this.connection, 'worker');
-
-        this.connection.subscribe('update-user-balance', {
-            callback: (error, msg) => this.callbackSubscribe(error, msg)
-    });
-
         try {
             await this.ipfsClient.createClient()
         } catch (e) {
@@ -244,7 +239,7 @@ export class Worker extends NatsService {
                     balance,
                     unit: 'Hbar',
                     operatorAccountId
-                });
+                }, false);
             } catch (error) {
                 throw new Error(`Worker (${['api-gateway', 'update-user-balance'].join('.')}) send: ` + error);
             }

worker-service/src/app.ts

@@ -36,6 +36,19 @@ Promise.all([
 
     const logger: PinoLogger = pinoLoggerInitialization(loggerMongo);
 
+    setInterval(() => {
+        const memoryUsage = process.memoryUsage();
+        const stack = (memoryUsage.rss - memoryUsage.heapTotal - memoryUsage.external) / 1024 / 1024;
+        console.log({
+            service: 'worker',
+            rss: `${(memoryUsage.rss / 1024 / 1024).toFixed(2)} MB`,
+            heapTotal: `${(memoryUsage.heapTotal / 1024 / 1024).toFixed(2)} MB`,
+            heapUsed: `${(memoryUsage.heapUsed / 1024 / 1024).toFixed(2)} MB`,
+            external: `${(memoryUsage.external / 1024 / 1024).toFixed(2)} MB`,
+            stack: `${stack.toFixed(2)} MB`
+        });
+    }, 10000);
+
     await new Users().setConnection(cn).init();
     const state = new ApplicationState();
     await state.setServiceName('WORKER').setConnection(cn).init();