From 9baec7a7cb062d4b6d076c606d73e578831a63ac Mon Sep 17 00:00:00 2001 From: andrei-mihnea-cerbu Date: Wed, 13 Nov 2024 17:23:21 +0200 Subject: [PATCH] network chunk v20 --- CEO/package-lock.json | 11 + CEO/package.json | 1 + CEO/src/helpers/announcement_sender.ts | 30 ++- CEO/src/helpers/backup_manager.ts | 31 ++- CEO/src/helpers/backup_retrieval.ts | 28 ++- CEO/src/helpers/department_sharer.ts | 43 +++- CEO/src/helpers/directory_watcher.ts | 4 + CEO/src/helpers/file_sharer.ts | 38 ++- CEO/src/helpers/json_manager.ts | 49 ++-- CEO/src/helpers/network_scanner.ts | 15 +- CEO/src/helpers/tcp_communicator.ts | 5 + CEO/src/helpers/users_info_fetcher.ts | 17 +- CEO/src/helpers/worker_manager.ts | 236 ++++++------------ CEO/src/network/operation_codes.ts | 3 +- .../operations_custom/general_operations.ts | 6 +- .../socket_communicator_base.ts | 42 +--- .../tcp_client_communicator.ts | 18 +- .../tcp_server_communicator.ts | 16 +- CEO/src/network/udp/udp_client.ts | 7 +- CEO/src/workers/backup_retrieval_worker.ts | 54 ++-- CEO/src/workers/directories_watcher_worker.ts | 39 ++- CEO/src/workers/network_scanner_worker.ts | 50 ++-- .../workers/resource_coordinator_worker.ts | 35 ++- CEO/src/workers/send_announcement_worker.ts | 40 +-- CEO/src/workers/servers_worker.ts | 36 ++- .../operations_custom/general_operations.ts | 6 +- User/src/helpers/backup_manager.ts | 22 +- User/src/helpers/backup_retrieval.ts | 24 +- User/src/helpers/department_sharer.ts | 22 +- User/src/helpers/file_sharer.ts | 22 +- User/src/helpers/network_scanner.ts | 4 +- User/src/helpers/users_info_fetcher.ts | 13 +- User/src/helpers/worker_manager.ts | 20 +- User/src/network/operation_codes.ts | 3 +- .../operations_custom/general_operations.ts | 6 +- User/src/network/udp/udp_client.ts | 7 +- User/src/workers/backup_retrieval_worker.ts | 24 +- .../src/workers/directories_watcher_worker.ts | 19 ++ User/src/workers/network_scanner_worker.ts | 17 ++ .../workers/resource_coordinator_worker.ts | 20 ++ User/src/workers/servers_worker.ts | 23 +- 41 files changed, 714 insertions(+), 392 deletions(-) diff --git a/CEO/package-lock.json b/CEO/package-lock.json index 4cb9f9f..05e8e3b 100644 --- a/CEO/package-lock.json +++ b/CEO/package-lock.json @@ -23,6 +23,7 @@ "@types/ping": "^0.4.4", "@types/proper-lockfile": "^4.1.4", "@types/uuid": "^10.0.0", + "check-disk-space": "^3.4.0", "copyfiles": "^2.4.1", "del-cli": "^5.0.0", "electron": "^33.0.2", @@ -1489,6 +1490,16 @@ "url": "https://github.com/chalk/chalk?sponsor=1" } }, + "node_modules/check-disk-space": { + "version": "3.4.0", + "resolved": "https://registry.npmjs.org/check-disk-space/-/check-disk-space-3.4.0.tgz", + "integrity": "sha512-drVkSqfwA+TvuEhFipiR1OC9boEGZL5RrWvVsOthdcvQNXyCCuKkEiTOTXZ7qxSf/GLwq4GvzfrQD/Wz325hgw==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=16" + } + }, "node_modules/chownr": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/chownr/-/chownr-2.0.0.tgz", diff --git a/CEO/package.json b/CEO/package.json index 6ce7ebe..cfcf5fe 100644 --- a/CEO/package.json +++ b/CEO/package.json @@ -29,6 +29,7 @@ "@types/ping": "^0.4.4", "@types/proper-lockfile": "^4.1.4", "@types/uuid": "^10.0.0", + "check-disk-space": "^3.4.0", "copyfiles": "^2.4.1", "del-cli": "^5.0.0", "electron": "^33.0.2", diff --git a/CEO/src/helpers/announcement_sender.ts b/CEO/src/helpers/announcement_sender.ts index c1f1fd5..c23640d 100644 --- a/CEO/src/helpers/announcement_sender.ts +++ b/CEO/src/helpers/announcement_sender.ts @@ -2,13 +2,14 @@ import { JsonManager } from './json_manager'; // Assuming this manages JSON conf import { TcpCommunicator } from "./tcp_communicator"; import { operationCodes } from '../network/operation_codes'; // Assuming this holds operation codes import { parentPort } from 'worker_threads'; -import {ParsedMessage} from "../network/message_handler"; +import { ParsedMessage } from "../network/message_handler"; export class AnnouncementSender { private applicationInfo: JsonManager; private readonly clientPort: number; private message: string = ''; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; constructor(applicationInfoPath: string, clientPort: number) { this.applicationInfo = new JsonManager(applicationInfoPath); @@ -18,6 +19,8 @@ export class AnnouncementSender { async start(message: string): Promise { console.log('AnnouncementWorker started.'); this.message = message; + this.stopRequested = false; + try { const activeUsersIp = await this.applicationInfo.readValue('users_ip'); if (!activeUsersIp || !activeUsersIp.length) { @@ -25,6 +28,11 @@ export class AnnouncementSender { } for (const ip of activeUsersIp) { + if (this.stopRequested) { + console.log('AnnouncementWorker stopped.'); + break; + } + const success = await this.sendAnnouncementToIp(ip); if (!success) { throw new Error(`Failed to send announcement to all users.`); @@ -32,15 +40,24 @@ export class AnnouncementSender { console.log(`Announcement sent and confirmed successfully from ${ip}`); } - parentPort?.postMessage({ success: true, message: 'Announcement sent to all active users successfully.' }); + process.send?.({ type: 'showAlert', message: 'Announcement sent to all active users successfully.' }); } catch (error: any) { console.error('Error in AnnouncementWorker:', error); - parentPort?.postMessage({ success: false, message: `A problem occurred: ${error.message}` }); + process.send?.({ type: 'shotAlert', message: `A problem occurred: ${error.message}` }); } console.log('AnnouncementWorker finished.'); } + async stop(): Promise { + console.log('Stopping AnnouncementWorker...'); + this.stopRequested = true; + if (this.tcpCommunicator) { + await this.tcpCommunicator.disconnect(); + } + console.log('AnnouncementWorker stopped.'); + } + private async sendAnnouncementToIp(ip: string): Promise { this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort); if (!await this.tcpCommunicator.connect()) { @@ -73,7 +90,12 @@ export class AnnouncementSender { private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { - if (!this.tcpCommunicator) return null; + if (this.stopRequested || !this.tcpCommunicator) { + clearInterval(idResponseCheck); + resolve(null); + return; + } + if (this.tcpCommunicator.hasResponseArrived()) { clearInterval(idResponseCheck); resolve(this.tcpCommunicator.getLastResult()); diff --git a/CEO/src/helpers/backup_manager.ts b/CEO/src/helpers/backup_manager.ts index f01aa67..ae87b65 100644 --- a/CEO/src/helpers/backup_manager.ts +++ b/CEO/src/helpers/backup_manager.ts @@ -5,7 +5,6 @@ import { MemoryManager } from './memory_manager'; import { JsonManager } from './json_manager'; import { TcpCommunicator } from './tcp_communicator'; import { operationCodes } from '../network/operation_codes'; -import { parentPort } from 'worker_threads'; import { ParsedMessage } from "../network/message_handler"; export class BackupManager { @@ -15,6 +14,8 @@ export class BackupManager { private userConfig: JsonManager; private readonly clientPort: number; private isBusy: boolean = false; + private intervalId: NodeJS.Timeout | null = null; + private stopRequested: boolean = false; constructor(userConfigPath: string, applicationInfoPath: string, memoryManagerPath: string, clientPort: number) { this.applicationInfo = new JsonManager(applicationInfoPath); @@ -24,12 +25,16 @@ export class BackupManager { } async start(): Promise { - setInterval(async () => { - if (!this.isBusy) { + this.intervalId = setInterval(async () => { + if (!this.isBusy || !this.stopRequested) { this.isBusy = true; this.log('Start successfully. Backup files to users.'); await this.initialize(); } + + if (global.gc) { + global.gc(); + } }, 10000); // 10-second interval for testing } @@ -139,9 +144,9 @@ export class BackupManager { } if (unsentFiles.length > 0) { - parentPort?.postMessage({ success: false, message: 'Backup could not be completed for all files', unsentFiles }); + process.send?.({type: 'log', message: 'Backup could not be completed for all files'}); } else { - parentPort?.postMessage({ success: true, message: 'Backup completed successfully' }); + process.send?.({type: 'log', message: 'Backup completed successfully' }); } } @@ -157,6 +162,22 @@ export class BackupManager { }); } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + // Unified logging function private log(message: string, level: 'log' | 'error' | 'warn' = 'log'): void { const prefix = '[BackupManager]'; diff --git a/CEO/src/helpers/backup_retrieval.ts b/CEO/src/helpers/backup_retrieval.ts index 2336b91..aec9696 100644 --- a/CEO/src/helpers/backup_retrieval.ts +++ b/CEO/src/helpers/backup_retrieval.ts @@ -1,7 +1,6 @@ import { JsonManager } from './json_manager'; import { TcpCommunicator } from "./tcp_communicator"; import { operationCodes } from '../network/operation_codes'; -import { parentPort } from 'worker_threads'; import path from 'path'; import fs from 'fs'; import crypto from 'crypto'; @@ -15,6 +14,8 @@ export class BackupRetrievalWorker { private encryptionKey: Buffer | null = null; private iv: Buffer | null = null; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; + private isBusy: boolean = false; constructor(userConfigPath: string, applicationInfoPath: string, clientPort: number, destinationPath: string) { this.userConfig = new JsonManager(userConfigPath); @@ -34,6 +35,8 @@ export class BackupRetrievalWorker { } async start(): Promise { + if(!this.stopRequested) return; + this.isBusy = true; try { const userInfo = await this.userConfig.readValue('user_info'); if (!userInfo || !userInfo.name) { @@ -62,13 +65,19 @@ export class BackupRetrievalWorker { this.log(`Backup retrieved successfully from ${ip}`); } - parentPort?.postMessage({ success: true, message: 'Backup retrieval completed successfully.' }); + process.send?.({ type: 'showAlert', message: 'Backup retrieval completed successfully.' }); + process.send?.({ type: 'changeContent', page: 'main_menu' }); } catch (error: any) { this.log(`Error in BackupRetrievalWorker: ${error.message}`, 'error'); - parentPort?.postMessage({ success: false, message: `A problem occurred: ${error.message}` }); + process.send?.({ type: 'showAlert', message: `A problem occurred: ${error.message}` }); + process.send?.({ type: 'changeContent', page: 'main_menu' }); + } + finally { + this.isBusy = false; } - finally{ + if (global.gc) { + global.gc(); } } @@ -172,6 +181,17 @@ export class BackupRetrievalWorker { } } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { diff --git a/CEO/src/helpers/department_sharer.ts b/CEO/src/helpers/department_sharer.ts index 62dc6b5..ea844b7 100644 --- a/CEO/src/helpers/department_sharer.ts +++ b/CEO/src/helpers/department_sharer.ts @@ -14,6 +14,8 @@ export class DepartmentSharer { private readonly clientPort: number; private isBusy: boolean = false; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; + private intervalId: NodeJS.Timeout | null = null; constructor( userConfigPath: string, @@ -30,12 +32,16 @@ export class DepartmentSharer { // Start sharing files with the department every minute async start(): Promise { - setInterval(async () => { - if (!this.isBusy) { + this.intervalId = setInterval(async () => { + if (!this.isBusy || !this.stopRequested) { this.isBusy = true; this.log('Start successfully. Sharing files with the department.'); await this.shareFilesWithDepartment(); } + + if (global.gc) { + global.gc(); + } }, 10000); // 10-second interval for testing } @@ -87,11 +93,10 @@ export class DepartmentSharer { const userIp = user.ip; this.tcpCommunicator = new TcpCommunicator(userIp, this.clientPort); - if (await this.tcpCommunicator.connect()) continue; + if (!await this.tcpCommunicator.connect()) continue; - // First clear the department directory - const clearSuccess = await this.clearDepartmentDirectory(); - if (clearSuccess) { + // First clear the department directory; + if (await this.clearDepartmentDirectory(userName)) { await this.sendFilesToUser(departmentFiles.structure, userName); } @@ -108,10 +113,10 @@ export class DepartmentSharer { } // Clear the department directory for a user - private async clearDepartmentDirectory(): Promise { + private async clearDepartmentDirectory(userName: string): Promise { if(!this.tcpCommunicator) return false; - if (!await this.tcpCommunicator.sendMessage(operationCodes.CLEAR_DEPARTMENT)) return false; + if (!await this.tcpCommunicator.sendMessage(operationCodes.CLEAR_DEPARTMENT, {userName: userName})) return false; const response = await this.waitForResponse(); if (!response || response.operationCode !== operationCodes.OK){ @@ -127,6 +132,8 @@ export class DepartmentSharer { if(!this.tcpCommunicator) return; const unsentFiles = Object.keys(files); + console.log(`\n\n${unsentFiles}\n\n`); + for (const fileName of unsentFiles) { const filePath = files[fileName]; @@ -152,17 +159,37 @@ export class DepartmentSharer { // Send the file if (!await this.tcpCommunicator.sendMessage(operationCodes.DEPARTMENT_FILE, metaInfo, Buffer.from(fileContent))) return; + console.log(`\n\nSending file: ${fileName} to ${userName}\n\n`); + const response = await this.waitForResponse(); if (!response || response.operationCode !== operationCodes.OK) { this.log(`Failed to send file: ${fileName}`, 'error'); return; } + this.log(`File sent successfully: ${fileName} to ${userName}`); + unsentFiles.splice(unsentFiles.indexOf(fileName), 1); await this.tcpCommunicator.disconnect(); } } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { diff --git a/CEO/src/helpers/directory_watcher.ts b/CEO/src/helpers/directory_watcher.ts index a96beaf..7549485 100644 --- a/CEO/src/helpers/directory_watcher.ts +++ b/CEO/src/helpers/directory_watcher.ts @@ -140,6 +140,10 @@ export class DirectoryWatcher { this.directoryWatcher.close(); this.directoryWatcher = null; } + + if (global.gc) { + global.gc(); + } } // Unified logging function diff --git a/CEO/src/helpers/file_sharer.ts b/CEO/src/helpers/file_sharer.ts index 68e254d..cada88f 100644 --- a/CEO/src/helpers/file_sharer.ts +++ b/CEO/src/helpers/file_sharer.ts @@ -17,6 +17,8 @@ export class FileSharer { private readonly clientPort: number; private isBusy: boolean; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; + private intervalId: NodeJS.Timeout | null = null; constructor(queueFilePath: string, clientPort: number) { this.queueManager = new QueueManager(queueFilePath, compareFnFileItemTask); @@ -26,26 +28,24 @@ export class FileSharer { // Start processing the file queue async start(): Promise { - setInterval(async () => { - if (!this.isBusy) { // Check if the queue is already being processed + this.intervalId = setInterval(async () => { + if (!this.isBusy || !this.stopRequested) { // Check if the queue is already being processed + this.isBusy = true; // Set busy flag to true before starting this.log("Start successfully. Processing the queue."); - await this.processQueue(); // Process the queue at regular intervals - this.log("Queue processing completed."); + await this.processQueue(); + } + + if (global.gc) { + global.gc(); } }, 10000); // 10 seconds interval } // Method to process the queue private async processQueue(): Promise { - if (this.isBusy) { - this.log("Queue is already being processed. Skipping this interval."); - return; - } - - this.isBusy = true; // Set busy flag to true before starting - while (!this.queueManager.isEmpty()) { const task = this.queueManager.peek(); + this.log('trimiti fisier'); if (task) { this.log(`Processing task for file: ${task.path} to IP: ${task.ip}`); @@ -106,6 +106,22 @@ export class FileSharer { return true; } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { diff --git a/CEO/src/helpers/json_manager.ts b/CEO/src/helpers/json_manager.ts index 5ed1db3..627acac 100644 --- a/CEO/src/helpers/json_manager.ts +++ b/CEO/src/helpers/json_manager.ts @@ -1,46 +1,47 @@ import fs from 'fs'; -import * as lockfile from 'proper-lockfile'; import path from 'path'; export class JsonManager { private readonly filePath: string; + private readonly lockFilePath: string; constructor(filePath: string) { const dir = path.dirname(filePath); - // Check if the directory exists, throw an error if it doesn't + // Check if the directory exists, throw error if it doesn't if (!fs.existsSync(dir)) { throw new Error(`The directory does not exist: ${dir}`); } this.filePath = filePath; + this.lockFilePath = `${filePath}.lock`; // Define the lock file path - // If the file doesn't exist, create it with an empty JSON object + // If the file doesn't exist, create it if (!fs.existsSync(filePath)) { fs.writeFileSync(filePath, JSON.stringify({}, null, 2), 'utf8'); } } - // Method to acquire a lock with retries - private async acquireLock(): Promise<() => Promise> { - return lockfile.lock(this.filePath, { - retries: { - retries: 20, // Retry up to 10 times - factor: 1, // Retry factor - minTimeout: 100, // Minimum delay between retries in ms - maxTimeout: 200 // Maximum delay between retries in ms - } - }); + // Method to acquire a lock (create .lock file) + private async acquireLock(): Promise { + while (fs.existsSync(this.lockFilePath)) { + // Wait until the lock file is released + await new Promise(resolve => setTimeout(resolve, 100)); // 100ms delay before retrying + } + // Create the lock file + fs.writeFileSync(this.lockFilePath, ''); } - // Method to release the lock - private async releaseLock(release: () => Promise): Promise { - await release(); + // Method to release the lock (delete .lock file) + private releaseLock(): void { + if (fs.existsSync(this.lockFilePath)) { + fs.unlinkSync(this.lockFilePath); + } } // Read a value by key from the JSON file with a lock public async readValue(key: string): Promise { - const release = await this.acquireLock(); + await this.acquireLock(); // Acquire the lock try { if (!fs.existsSync(this.filePath)) return null; @@ -51,13 +52,13 @@ export class JsonManager { console.error(`Error reading from JSON file: ${err.message}`); return null; } finally { - await this.releaseLock(release); // Always release the lock after the operation + this.releaseLock(); // Always release the lock after the operation } } // Write a key-value pair to the JSON file with a lock public async writeValue(key: string, value: any): Promise { - const release = await this.acquireLock(); + await this.acquireLock(); // Acquire the lock try { let data: { [key: string]: any } = {}; @@ -75,13 +76,13 @@ export class JsonManager { console.error(`Error writing to JSON file: ${err.message}`); return false; } finally { - await this.releaseLock(release); // Always release the lock after the operation + this.releaseLock(); // Always release the lock after the operation } } // Remove a key-value pair from the JSON file with a lock public async removeValue(key: string): Promise { - const release = await this.acquireLock(); + await this.acquireLock(); // Acquire the lock try { if (!fs.existsSync(this.filePath)) return false; @@ -97,13 +98,13 @@ export class JsonManager { console.error(`Error removing key from JSON file: ${err.message}`); return false; } finally { - await this.releaseLock(release); // Always release the lock after the operation + this.releaseLock(); // Always release the lock after the operation } } // Reset the JSON file by clearing all data with a lock public async resetFile(): Promise { - const release = await this.acquireLock(); + await this.acquireLock(); // Acquire the lock try { fs.writeFileSync(this.filePath, JSON.stringify({}, null, 2), 'utf8'); @@ -112,7 +113,7 @@ export class JsonManager { console.error(`Error resetting JSON file: ${err.message}`); return false; } finally { - await this.releaseLock(release); // Always release the lock after the operation + this.releaseLock(); // Always release the lock after the operation } } } diff --git a/CEO/src/helpers/network_scanner.ts b/CEO/src/helpers/network_scanner.ts index 238ef3c..d1ce897 100644 --- a/CEO/src/helpers/network_scanner.ts +++ b/CEO/src/helpers/network_scanner.ts @@ -1,6 +1,5 @@ import {JsonManager} from "./json_manager"; import {UdpClient} from "../network/udp/udp_client"; -import {parentPort} from "worker_threads"; import {TcpCommunicator} from "./tcp_communicator"; import {operationCodes} from "../network/operation_codes"; import {ParsedMessage} from "../network/message_handler"; @@ -55,7 +54,7 @@ export class NetworkScanner { try { this.log('UC Check running...', 'log', 'startUCCheck'); const udpClient = new UdpClient(this.udpPort); - const aliveClients = await udpClient.getAliveClients(); + const aliveClients = await udpClient.getTargetClients(operationCodes.ARE_YOU_UC); const storedIp = await this.applicationInfo.readValue('serverIp'); const foundClient = aliveClients.length > 0; @@ -65,19 +64,19 @@ export class NetworkScanner { if (!storedIp || storedIp !== ipAddress) { await this.applicationInfo.writeValue('serverIp', ipAddress); if (!this.appStarted) { - parentPort?.postMessage({ type: 'changeContent', page: this.okPage }); + process.send?.({ type: 'changeContent', page: this.okPage }); } this.appStarted = true; } else if (!this.appStarted) { - parentPort?.postMessage({ type: 'changeContent', page: this.okPage }); + process.send?.({ type: 'changeContent', page: this.okPage }); this.appStarted = true; } } else { - parentPort?.postMessage({ type: 'changeContent', page: this.errorPage }); + process.send?.({ type: 'changeContent', page: this.errorPage }); } } catch (err) { this.log(`Error checking UC: ${err}`, 'error', 'startUCCheck'); - parentPort?.postMessage({ type: 'changeContent', page: this.errorPage }); + process.send?.({ type: 'changeContent', page: this.errorPage }); } finally { this.ucCheckBusy = false; } @@ -96,7 +95,7 @@ export class NetworkScanner { this.log('IP Lookup running...', 'log', 'startUserIPLookup'); const serverIp = await this.applicationInfo.readValue('serverIp'); const udpClient = new UdpClient(this.udpPort); - const activeIPs = await udpClient.getAliveClients(); + const activeIPs = await udpClient.getTargetClients(operationCodes.ARE_YOU_HUMAN); const filteredIPs = activeIPs.filter(ip => ip !== serverIp); // Save the filtered IPs to 'users_ip' @@ -152,7 +151,7 @@ export class NetworkScanner { if (response?.operationCode !== operationCodes.OK) { await this.userConfig.resetFile(); await this.userConfig.writeValue('app_type', app_type); - parentPort?.postMessage({ type: 'changeContent', page: this.databaseResetPage }); + process.send?.({ type: 'changeContent', page: this.databaseResetPage }); } } catch (err) { this.log(`Error during login request: ${err}`, 'error', 'sendLoginRequest'); diff --git a/CEO/src/helpers/tcp_communicator.ts b/CEO/src/helpers/tcp_communicator.ts index ac85298..648b71a 100644 --- a/CEO/src/helpers/tcp_communicator.ts +++ b/CEO/src/helpers/tcp_communicator.ts @@ -43,6 +43,11 @@ export class TcpCommunicator { getLastResult(): ParsedMessage | null { const message = this.lastResult; this.lastResult = null; + + if (global.gc) { + global.gc(); + } + return message; } diff --git a/CEO/src/helpers/users_info_fetcher.ts b/CEO/src/helpers/users_info_fetcher.ts index 20c0b24..2852b04 100644 --- a/CEO/src/helpers/users_info_fetcher.ts +++ b/CEO/src/helpers/users_info_fetcher.ts @@ -11,6 +11,7 @@ export class UsersInfoFetcher { private readonly clientPort: number; private memoryId: string; private readonly activeUsersKey: string; + private intervalId: NodeJS.Timeout | null = null; constructor(applicationInfoPath: string, memoryManagerPath: string, clientPort: number) { this.applicationInfo = new JsonManager(applicationInfoPath); @@ -23,9 +24,13 @@ export class UsersInfoFetcher { // Method to start checking user info periodically (every minute) async start(): Promise { - setInterval(async () => { + this.intervalId = setInterval(async () => { await this.initialize(); // Re-run every minute - }, 5000); // 1 minute interval + + if (global.gc) { + global.gc(); + } + }, 5000); // 5-second interval for testing } // Initialize and fetch user IPs and process users info @@ -97,6 +102,14 @@ export class UsersInfoFetcher { await this.memoryManager.updateMetaInformation(this.memoryId, userInfo); // Update active users in memory } + stop(): void { + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + console.log("[UsersInfoFetcher] Stopped successfully."); + } + } + // Unified logging function private log(message: string, level: 'log' | 'error' = 'log'): void { const prefix = '[UsersInfoFetcher]'; diff --git a/CEO/src/helpers/worker_manager.ts b/CEO/src/helpers/worker_manager.ts index a2401bc..7a9ae9f 100644 --- a/CEO/src/helpers/worker_manager.ts +++ b/CEO/src/helpers/worker_manager.ts @@ -1,214 +1,128 @@ -import { Worker } from 'worker_threads'; +import { fork, ChildProcess } from 'child_process'; import path from 'path'; -import {WindowManager} from "./window_manager"; +import { WindowManager } from "./window_manager"; export class WorkerManager { private readonly pathToWorkerDir: string; - private windowManager: WindowManager - private workers: Worker[]; // Array to store running workers + private windowManager: WindowManager; + private workers: ChildProcess[]; + private cleanupInProgress: boolean = false; constructor(pathToWorkerDir: string, windowManager: WindowManager) { this.pathToWorkerDir = pathToWorkerDir; this.windowManager = windowManager; - this.workers = []; // Initialize the array to store workers + this.workers = []; } async startNetworkScannerWorker(udpPort: number, tcpPort: number, okPage: string, errorPage: string, databaseResetPage: string, userConfigPath: string, applicationInfoPath: string): Promise { - return new Promise((resolve, reject) => { - const worker = new Worker(path.join(this.pathToWorkerDir, 'network_scanner_worker.js'), { - workerData: { udpPort, tcpPort, okPage, errorPage, databaseResetPage, userConfigPath, applicationInfoPath }, // Pass necessary data to the worker - }); - - this.workers.push(worker); // Store the worker reference - - worker.on('message', (data) => { - if (data.type === 'changeContent') { - this.windowManager.changeContent(data.page); - } - }); - - worker.on('error', (err) => { - console.error('Network Scanner Worker error:', err); - worker.terminate(); - this.removeWorker(worker); - reject(err); // Reject the promise if there's an error - }); - - worker.on('exit', (code) => { - console.log(`Network Scanner Worker exited with code ${code}`); - this.removeWorker(worker); // Remove worker reference when it exits - resolve(); // Resolve when the worker exits cleanly - }); + return this.startForkedWorker('network_scanner_worker.js', { + UDP_PORT: udpPort.toString(), + TCP_PORT: tcpPort.toString(), + OK_PAGE: okPage, + ERROR_PAGE: errorPage, + DATABASE_RESET_PAGE: databaseResetPage, + USER_CONFIG_PATH: userConfigPath, + APPLICATION_INFO_PATH: applicationInfoPath }); } - // Start the Directories Watcher Worker async startDirectoriesWatchersWorker(memoryManagerPath: string, applicationInfoPath: string): Promise { - return new Promise((resolve, reject) => { - const worker = new Worker(path.join(this.pathToWorkerDir, 'directories_watcher_worker.js'), { - workerData: { memoryManagerPath, applicationInfoPath }, // Pass the port to the worker - }); - - this.workers.push(worker); // Store the worker reference - - worker.on('message', (data) => { - console.log('DirectoriesWatcher message:', data); - }); - - worker.on('error', (err) => { - console.error('DirectoriesWatcher error:', err); - worker.terminate(); - this.removeWorker(worker); - reject(err); // Reject the promise if there's an error - }); - - worker.on('exit', (code) => { - console.log(`DirectoriesWatcher exited with code ${code}`); - this.removeWorker(worker); // Remove worker reference when it exits - resolve(); // Resolve when the worker exits cleanly - }); + return this.startForkedWorker('directories_watcher_worker.js', { + MEMORY_MANAGER_PATH: memoryManagerPath, + APPLICATION_INFO_PATH: applicationInfoPath }); } - // Start the Servers Worker (UDP and TCP servers) async startServersWorker(host: string, udpPort: number, tcpPort: number): Promise { - return new Promise((resolve, reject) => { - const worker = new Worker(path.join(this.pathToWorkerDir, 'servers_worker.js'), { - workerData: { HOST: host, USER_UDP_PORT: udpPort, USER_TCP_PORT: tcpPort }, // Pass host and ports to the worker - }); - - this.workers.push(worker); // Store the worker reference - - worker.on('message', (data) => { - console.log('Servers Worker message:', data); - }); - - worker.on('error', (err) => { - console.error('Servers Worker error:', err); - worker.terminate(); - this.removeWorker(worker); - reject(err); // Reject the promise if there's an error - }); - - worker.on('exit', (code) => { - console.log(`Servers Worker exited with code ${code}`); - this.removeWorker(worker); // Remove worker reference when it exits - resolve(); // Resolve when the worker exits cleanly - }); + return this.startForkedWorker('servers_worker.js', { + HOST: host, + USER_UDP_PORT: udpPort.toString(), + USER_TCP_PORT: tcpPort.toString() }); } - // Start the Users Info Worker async startResourceCoordinatorWorker(usersConfigPath: string, applicationInfoPath: string, memoryManagerPath: string, queueManagerPath: string, tcpPort: number): Promise { - return new Promise((resolve, reject) => { - const worker = new Worker(path.join(this.pathToWorkerDir, 'resource_coordinator_worker.js'), { - workerData: { - usersConfigPath, - applicationInfoPath, - memoryManagerPath, - queueManagerPath, - tcpPort - }, // Pass necessary parameters to the worker - }); - - this.workers.push(worker); // Store the worker reference - - worker.on('message', (data) => { - console.log('Users Info Worker message:', data); - }); - - worker.on('error', (err) => { - console.error('Users Info Worker error:', err); - worker.terminate(); - this.removeWorker(worker); - reject(err); // Reject the promise if there's an error - }); - - worker.on('exit', (code) => { - console.log(`Users Info Worker exited with code ${code}`); - this.removeWorker(worker); // Remove worker reference when it exits - resolve(); // Resolve when the worker exits cleanly - }); + return this.startForkedWorker('resource_coordinator_worker.js', { + USERS_CONFIG_PATH: usersConfigPath, + APPLICATION_INFO_PATH: applicationInfoPath, + MEMORY_MANAGER_PATH: memoryManagerPath, + QUEUE_MANAGER_PATH: queueManagerPath, + TCP_PORT: tcpPort.toString() }); } - // Start the Backup Retrieval Worker - async startBackupRetrievalWorker( - userConfigPath: string, - applicationInfoPath: string, - clientPort: number, - destinationPath: string - ): Promise { - return new Promise(async (resolve, reject) => { - const worker = new Worker(path.join(this.pathToWorkerDir, 'backup_retrieval_worker.js'), { - workerData: { userConfigPath, applicationInfoPath, clientPort, destinationPath }, // Pass parameters to the worker - }); - - this.workers.push(worker); // Store the worker reference - - worker.on('message', async (data) => { - console.log('Backup Retrieval Worker message:', data); - await new Promise((resolve) => setTimeout(resolve, 3000)); // Wait for 1 second - await this.windowManager.changeContent('main_menu'); - await this.windowManager.showAlert(data.message); - }); - - worker.on('error', (err) => { - console.error('Backup Retrieval Worker error:', err); - worker.terminate(); - this.removeWorker(worker); - reject(err); // Reject the promise if there's an error - }); - - worker.on('exit', (code) => { - console.log(`Backup Retrieval Worker exited with code ${code}`); - this.removeWorker(worker); // Remove worker reference when it exits - resolve(); // Resolve when the worker exits cleanly - }); + async startBackupRetrievalWorker(userConfigPath: string, applicationInfoPath: string, clientPort: number, destinationPath: string): Promise { + return this.startForkedWorker('backup_retrieval_worker.js', { + USER_CONFIG_PATH: userConfigPath, + APPLICATION_INFO_PATH: applicationInfoPath, + CLIENT_PORT: clientPort.toString(), + DESTINATION_PATH: destinationPath }); } async startAnnouncementWorker(applicationInfoPath: string, clientPort: number, message: string): Promise { + return this.startForkedWorker('send_announcement_worker.js', { + APPLICATION_INFO_PATH: applicationInfoPath, + CLIENT_PORT: clientPort.toString(), + MESSAGE: message + }); + } + + private async startForkedWorker(scriptName: string, envData: { [key: string]: string }): Promise { return new Promise((resolve, reject) => { - const worker = new Worker(path.join(this.pathToWorkerDir, 'send_announcement_worker.js'), { - workerData: { applicationInfoPath, clientPort, message }, // Pass necessary data to the worker + const worker = fork(path.join(this.pathToWorkerDir, scriptName), { + execArgv: ['--max-old-space-size=4096'], + env: { ...process.env, ...envData } }); - this.workers.push(worker); // Store the worker reference + this.workers.push(worker); - worker.on('message', async (data) => { - await new Promise((resolve) => setTimeout(resolve, 3000)); // Wait for 1 second - this.windowManager.changeContent('main_menu'); - this.windowManager.showAlert(`${data.message}`) + worker.on('message', (data: unknown) => { + const message = data as { type: string, page?: string, message?: string }; + + if (message.type === 'changeContent' && message.page) { + this.windowManager.changeContent(message.page); + } else if (message.type === 'showAlert' && message.message) { + this.windowManager.showAlert(message.message); + } else { + console.log(`${scriptName} message:`, message); + } }); worker.on('error', (err) => { - console.error('Announcement Worker error:', err); - worker.terminate(); + console.error(`${scriptName} error:`, err); + worker.kill(); this.removeWorker(worker); - reject(err); // Reject the promise if there's an error + reject(err); }); - worker.on('exit', (code) => { - console.log(`Announcement Worker exited with code ${code}`); - this.removeWorker(worker); // Remove worker reference when it exits - resolve(); // Resolve when the worker exits cleanly + worker.on('exit', (code, signal) => { + this.removeWorker(worker); + if (code === 0) { + console.log(`${scriptName} exited successfully`); + resolve(); + } else if (signal) { + console.log(`${scriptName} was killed with signal: ${signal}`); + } else { + console.error(`${scriptName} exited with code: ${code}`); + } }); }); } - // Close all running workers closeAllWorkers(): void { + if (this.cleanupInProgress) return; + this.cleanupInProgress = true; + console.log('Terminating all running workers...'); - this.workers.forEach(worker => worker.terminate()); // Terminate each worker - this.workers = []; // Clear the array after terminating all workers + this.workers.forEach(worker => worker.kill()); + this.workers = []; } - // Helper method to remove a worker from the workers array when it exits - private removeWorker(worker: Worker): void { + private removeWorker(worker: ChildProcess): void { const index = this.workers.indexOf(worker); if (index > -1) { - this.workers.splice(index, 1); // Remove the worker from the array + this.workers.splice(index, 1); } } } diff --git a/CEO/src/network/operation_codes.ts b/CEO/src/network/operation_codes.ts index 7f6edb1..fe7bd78 100644 --- a/CEO/src/network/operation_codes.ts +++ b/CEO/src/network/operation_codes.ts @@ -1,6 +1,7 @@ export let operationCodes = { // General Operations - HEARTBEAT: 'HEARTBEAT', + ARE_YOU_HUMAN: 'ARE_YOU_HUMAN', + ARE_YOU_UC: 'ARE_YOU_UC', ALIVE: 'ALIVE', SET_PUBLIC_KEY: 'SET_PUBLIC_KEY', SET_AES_KEY: 'SET_AES_KEY', diff --git a/CEO/src/network/operations_custom/general_operations.ts b/CEO/src/network/operations_custom/general_operations.ts index 4bdee66..afe4c50 100644 --- a/CEO/src/network/operations_custom/general_operations.ts +++ b/CEO/src/network/operations_custom/general_operations.ts @@ -7,14 +7,14 @@ export class GeneralOperations implements OperationPlugin { public static readonly operationCodes = { OK: 'OK', ERR: 'ERR', - HEARTBEAT: 'HEARTBEAT', + ARE_YOU_HUMAN: 'ARE_YOU_HUMAN', ALIVE: 'ALIVE', SET_PUBLIC_KEY: 'SET_PUBLIC_KEY', SET_AES_KEY: 'SET_AES_KEY', }; // Handle heartbeat operation asynchronously - public static async handleHeartbeat(): Promise { + public static async handleAreYouHuman(): Promise { const networkInterfaces = os.networkInterfaces(); let ipAddress = 'Unknown'; @@ -79,7 +79,7 @@ export class GeneralOperations implements OperationPlugin { // Register general operations with the OperationHandler public register(operationHandler: OperationHandler): void { - operationHandler.registerHandler(GeneralOperations.operationCodes.HEARTBEAT, GeneralOperations.handleHeartbeat); + operationHandler.registerHandler(GeneralOperations.operationCodes.ARE_YOU_HUMAN, GeneralOperations.handleAreYouHuman); operationHandler.registerHandler(GeneralOperations.operationCodes.SET_PUBLIC_KEY, GeneralOperations.handlePublicKey); operationHandler.registerHandler(GeneralOperations.operationCodes.SET_AES_KEY, GeneralOperations.handleAESKey); operationHandler.registerHandler(GeneralOperations.operationCodes.OK, GeneralOperations.handleOk); diff --git a/CEO/src/network/socket_communicator/socket_communicator_base.ts b/CEO/src/network/socket_communicator/socket_communicator_base.ts index 01eb64e..f749193 100644 --- a/CEO/src/network/socket_communicator/socket_communicator_base.ts +++ b/CEO/src/network/socket_communicator/socket_communicator_base.ts @@ -1,6 +1,5 @@ import { ParsedMessage } from '../message_handler'; import { OperationHandler } from '../operations_base/operation_handler'; -import ping from "ping"; import { constants, createCipheriv, @@ -16,16 +15,16 @@ export abstract class SocketCommunicatorBase { protected readonly port: number; protected readonly operationHandler: OperationHandler; protected handlerResult: ParsedMessage | null; - protected networkSpeed: number | null = null; protected chunkBuffers: { [messageId: string]: string[] }; - protected readonly EOP = ''; protected privateKey: string | null; protected publicKey: string | null; protected aesKey: Buffer | null; protected aesIv: Buffer | null; + protected readonly EOP = ''; + protected readonly CHUNK_SIZE = 1024; private incompleteChunkBuffer: string = ''; protected constructor(ip: string, port: number, operationHandler: OperationHandler) { @@ -113,40 +112,6 @@ export abstract class SocketCommunicatorBase { ).toString('base64'); } - protected async scanNetworkLatency(): Promise { - const targetIp = this.ip; // Use the IP from the superclass - - try { - const response = await ping.promise.probe(targetIp); - - if (!response.alive || response.time === "unknown") { - console.warn(`Ping failed to reach ${targetIp}. Using default network speed.`); - return 200; // Default latency in ms if ping fails - } - - return response.time; // Latency in ms from ping response - } catch (error: any) { - console.error(`Ping error: ${error.message}. Using default network speed.`); - return 200; // Default latency in ms if an error occurs - } - } - - // Calculate optimal chunk size based on network latency, with fallback if necessary - protected async calculateOptimalChunkSize(messageLength: number): Promise { - const latency = await this.scanNetworkLatency(); - this.networkSpeed = latency > 0 ? 1000 / latency : 1; // Speed in bytes/ms based on latency - - // Calculate initial chunk size based on latency (bounded between 512 and 1024 bytes) - let chunkSize = Math.min(Math.max(512, Math.floor(5000 / this.networkSpeed)), 1024); - - // Adjust chunk size for base64 alignment (multiple of 4) - while (messageLength % chunkSize !== 0 && chunkSize > 0) { - chunkSize -= 4; - } - - return chunkSize || 1024; // Fallback to 1024 if alignment adjustment results in 0 - } - async handleIncomingChunk(data: Buffer): Promise { // Append incoming data to the incomplete buffer this.incompleteChunkBuffer += data.toString(); @@ -171,9 +136,6 @@ export abstract class SocketCommunicatorBase { // Store the chunk in the correct position based on sequenceNumber (1-based indexing) this.chunkBuffers[header.messageId][header.sequenceNumber - 1] = chunkContent; - console.log(`Received chunk: ${incomingMessage}`); - console.log(`Chunks received so far: ${this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length}`); - // Check if all chunks have been received if (this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length === header.totalChunks) { // Join all chunks to form the full message diff --git a/CEO/src/network/socket_communicator/tcp_client_communicator.ts b/CEO/src/network/socket_communicator/tcp_client_communicator.ts index 4974773..11824b8 100644 --- a/CEO/src/network/socket_communicator/tcp_client_communicator.ts +++ b/CEO/src/network/socket_communicator/tcp_client_communicator.ts @@ -22,12 +22,10 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { } setServerPublicKey(publicKey: string): void { - console.log('\n\nSetting server public key\n\n'); this.publicKey = publicKey; } setAesKey(aesKey: string, aesIv: string): void { - console.log('\n\nSetting AES key\n\n'); this.aesKey = Buffer.from(aesKey, 'base64'); this.aesIv = Buffer.from(aesIv, 'base64'); } @@ -60,29 +58,29 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { } async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise { - if (!this.networkSpeed) { - this.networkSpeed = await this.scanNetworkLatency(); - } - const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); const outgoingMessage = this.encryptWithAes(message); // Calculate optimal chunk size based on network latency - const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length); - const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); + const totalChunks = Math.ceil(outgoingMessage.length / this.CHUNK_SIZE); + const messageId = Date.now().toString(); // Send each chunk with a delay between them for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); + const chunk = outgoingMessage.slice(i * this.CHUNK_SIZE, (i + 1) * this.CHUNK_SIZE); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, totalChunks, }); const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`; - if(!this.socket.write(chunkWithHeader)) this.socket.end(); + + if (!this.socket.write(chunkWithHeader)) { + // Wait for the 'drain' event before writing the next chunk + await new Promise((resolve) => this.socket.once('drain', resolve)); + } } } diff --git a/CEO/src/network/socket_communicator/tcp_server_communicator.ts b/CEO/src/network/socket_communicator/tcp_server_communicator.ts index 655c428..aa1779b 100644 --- a/CEO/src/network/socket_communicator/tcp_server_communicator.ts +++ b/CEO/src/network/socket_communicator/tcp_server_communicator.ts @@ -38,10 +38,6 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { } async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise { - if (!this.networkSpeed) { - this.networkSpeed = await this.scanNetworkLatency(); - } - const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); let outgoingMessage: string; @@ -56,21 +52,23 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { outgoingMessage = this.encryptWithAes(message); } - // Calculate optimal chunk size based on network latency - const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length); - const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); + const totalChunks = Math.ceil(outgoingMessage.length / this.CHUNK_SIZE); const messageId = Date.now().toString(); // Send each chunk with a delay between them for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); + const chunk = outgoingMessage.slice(i * this.CHUNK_SIZE, (i + 1) * this.CHUNK_SIZE); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, totalChunks, }); const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`; - if(!this.socket.write(chunkWithHeader)) this.socket.end(); + + if (!this.socket.write(chunkWithHeader)) { + // Wait for the 'drain' event before writing the next chunk + await new Promise((resolve) => this.socket.once('drain', resolve)); + } } } } diff --git a/CEO/src/network/udp/udp_client.ts b/CEO/src/network/udp/udp_client.ts index 2449639..a16c2e1 100644 --- a/CEO/src/network/udp/udp_client.ts +++ b/CEO/src/network/udp/udp_client.ts @@ -29,7 +29,7 @@ export class UdpClient { } // Send heartbeat and get clients that respond with ALIVE, excluding the host machine's IP addresses - async getAliveClients(): Promise { + async getTargetClients(heartbeatCode: string): Promise { const subnet = this.getSubnet(); const ipRange = this.getIPRange(subnet); @@ -45,7 +45,7 @@ export class UdpClient { const aliveClients: string[] = []; for (const ip of activeIps) { if (!localIPs.includes(ip)) { - const result = await this.sendHeartbeat(ip); + const result = await this.sendHeartbeat(ip, heartbeatCode); if (result.found) { aliveClients.push(ip); } @@ -73,9 +73,8 @@ export class UdpClient { } // Send heartbeat to an IP - private async sendHeartbeat(ip: string): Promise<{ found: boolean }> { + private async sendHeartbeat(ip: string, heartbeatCode: string): Promise<{ found: boolean }> { return new Promise((resolve) => { - const heartbeatCode = operationCodes.HEARTBEAT; const heartbeatMessage = MessageHandler.formatMessage(heartbeatCode); this.log(`Sending heartbeat to ${ip}`); diff --git a/CEO/src/workers/backup_retrieval_worker.ts b/CEO/src/workers/backup_retrieval_worker.ts index 337a4a6..ef1a3dd 100644 --- a/CEO/src/workers/backup_retrieval_worker.ts +++ b/CEO/src/workers/backup_retrieval_worker.ts @@ -1,18 +1,20 @@ -import { workerData, parentPort } from 'worker_threads'; -import { BackupRetrievalWorker} from '../helpers/backup_retrieval'; // Assuming the class is in the same folder +import { BackupRetrievalWorker } from '../helpers/backup_retrieval'; +import dotenv from 'dotenv'; -// Destructure the data passed from the WorkerManager -const { - userConfigPath, - applicationInfoPath, - clientPort, - destinationPath -}: { - userConfigPath: string, - applicationInfoPath: string, - clientPort: number, - destinationPath: string -} = workerData; +// Load environment variables from .env file if it exists +dotenv.config(); + +// Retrieve configuration from environment variables +const userConfigPath = process.env.USER_CONFIG_PATH as string; +const applicationInfoPath = process.env.APPLICATION_INFO_PATH as string; +const clientPort = Number(process.env.CLIENT_PORT); +const destinationPath = process.env.DESTINATION_PATH as string; + +// Validate that all required environment variables are present +if (!userConfigPath || !applicationInfoPath || !clientPort || !destinationPath) { + console.error('Error: Missing required environment variables.'); + process.exit(1); +} // Initialize the BackupRetrievalWorker const backupRetrievalWorker = new BackupRetrievalWorker( @@ -23,4 +25,26 @@ const backupRetrievalWorker = new BackupRetrievalWorker( ); // Start the backup retrieval process -backupRetrievalWorker.start(); +backupRetrievalWorker.start().then(() => { + console.log('Backup retrieval process completed successfully.'); +}); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + // Perform any cleanup, such as closing connections, saving data, etc. + // Example: if you have a server instance running, you may want to close it: + // await server.close(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} + diff --git a/CEO/src/workers/directories_watcher_worker.ts b/CEO/src/workers/directories_watcher_worker.ts index a5ba517..b864f40 100644 --- a/CEO/src/workers/directories_watcher_worker.ts +++ b/CEO/src/workers/directories_watcher_worker.ts @@ -1,11 +1,20 @@ -import {DirectoryWatcher} from "../helpers/directory_watcher"; -import {workerData} from "worker_threads"; +import { DirectoryWatcher } from "../helpers/directory_watcher"; +import dotenv from 'dotenv'; -const { - memoryManagerPath, - applicationInfoPath, -} = workerData; +// Load environment variables from .env file if it exists +dotenv.config(); +// Retrieve configuration from environment variables +const memoryManagerPath = process.env.MEMORY_MANAGER_PATH as string; +const applicationInfoPath = process.env.APPLICATION_INFO_PATH as string; + +// Validate that all required environment variables are present +if (!memoryManagerPath || !applicationInfoPath) { + console.error('Error: Missing required environment variables.'); + process.exit(1); +} + +// Initialize and start DirectoryWatcher instances const backupDirectoryManager = new DirectoryWatcher(memoryManagerPath, applicationInfoPath, 'backupDirectory'); backupDirectoryManager.start(); @@ -15,3 +24,21 @@ departmentShareManager.start(); const shareFileManager = new DirectoryWatcher(memoryManagerPath, applicationInfoPath, 'shareDirectory'); shareFileManager.start(); +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + backupDirectoryManager.closeWatcher(); + departmentShareManager.closeWatcher(); + shareFileManager.closeWatcher(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/CEO/src/workers/network_scanner_worker.ts b/CEO/src/workers/network_scanner_worker.ts index c22db64..f60bf03 100644 --- a/CEO/src/workers/network_scanner_worker.ts +++ b/CEO/src/workers/network_scanner_worker.ts @@ -1,19 +1,39 @@ -import { parentPort, workerData } from 'worker_threads'; +import { parentPort } from 'worker_threads'; import { NetworkScanner } from '../helpers/network_scanner'; -// Define the structure of workerData -interface WorkerData { - udpPort: number; - tcpPort: number; - okPage: string; - errorPage: string; - databaseResetPage: string; - userConfigPath: string; - applicationInfoPath: string; -} - -// Extract the data passed to the worker -const { udpPort, tcpPort, okPage, errorPage, databaseResetPage, userConfigPath, applicationInfoPath }: WorkerData = workerData; +// Extract data from environment variables +const udpPort = parseInt(process.env.UDP_PORT || '0', 10); +const tcpPort = parseInt(process.env.TCP_PORT || '0', 10); +const okPage = process.env.OK_PAGE || ''; +const errorPage = process.env.ERROR_PAGE || ''; +const databaseResetPage = process.env.DATABASE_RESET_PAGE || ''; +const userConfigPath = process.env.USER_CONFIG_PATH || ''; +const applicationInfoPath = process.env.APPLICATION_INFO_PATH || ''; // Start the NetworkScanner instance -const networkScanner = new NetworkScanner(applicationInfoPath, userConfigPath, udpPort, tcpPort, okPage, errorPage, databaseResetPage); +const networkScanner = new NetworkScanner( + applicationInfoPath, + userConfigPath, + udpPort, + tcpPort, + okPage, + errorPage, + databaseResetPage +); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + networkScanner.stopAllIntervals(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/CEO/src/workers/resource_coordinator_worker.ts b/CEO/src/workers/resource_coordinator_worker.ts index 1760ff0..c10513c 100644 --- a/CEO/src/workers/resource_coordinator_worker.ts +++ b/CEO/src/workers/resource_coordinator_worker.ts @@ -1,11 +1,14 @@ -import { workerData } from 'worker_threads'; import { UsersInfoFetcher } from '../helpers/users_info_fetcher'; -import {BackupManager} from "../helpers/backup_manager"; -import {FileSharer} from "../helpers/file_sharer"; -import {DepartmentSharer} from "../helpers/department_sharer"; +import { BackupManager } from '../helpers/backup_manager'; +import { FileSharer } from '../helpers/file_sharer'; +import { DepartmentSharer } from '../helpers/department_sharer'; -// Destructure the required information from workerData -const { usersConfigPath, applicationInfoPath, memoryManagerPath, queueManagerPath, tcpPort } = workerData; +// Retrieve data from environment variables +const usersConfigPath = process.env.USERS_CONFIG_PATH || ''; +const applicationInfoPath = process.env.APPLICATION_INFO_PATH || ''; +const memoryManagerPath = process.env.MEMORY_MANAGER_PATH || ''; +const queueManagerPath = process.env.QUEUE_MANAGER_PATH || ''; +const tcpPort = parseInt(process.env.TCP_PORT || '0', 10); const usersInfoFetcher = new UsersInfoFetcher(applicationInfoPath, memoryManagerPath, tcpPort); usersInfoFetcher.start(); @@ -16,7 +19,25 @@ backupManager.start(); const fileSharer = new FileSharer(queueManagerPath, tcpPort); fileSharer.start(); - const departmentSharer = new DepartmentSharer(usersConfigPath, applicationInfoPath, memoryManagerPath, tcpPort); departmentSharer.start(); +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + usersInfoFetcher.stop(); + await backupManager.stop(); + await fileSharer.stop(); + await departmentSharer.stop(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/CEO/src/workers/send_announcement_worker.ts b/CEO/src/workers/send_announcement_worker.ts index 816b6e7..2d772f2 100644 --- a/CEO/src/workers/send_announcement_worker.ts +++ b/CEO/src/workers/send_announcement_worker.ts @@ -1,19 +1,27 @@ -import { workerData } from 'worker_threads'; -import { AnnouncementSender} from "../helpers/announcement_sender"; +import { AnnouncementSender } from "../helpers/announcement_sender"; -// Destructure data passed from the main thread -const { - applicationInfoPath, - clientPort, - message -}: { - applicationInfoPath: string, - clientPort: number, - message: string -} = workerData; +// Read environment variables passed by WorkerManager +const applicationInfoPath = process.env.APPLICATION_INFO_PATH as string; +const clientPort = parseInt(process.env.CLIENT_PORT as string, 10); +const message = process.env.MESSAGE as string; -// Initialize the AnnouncementWorker -const announcementWorker = new AnnouncementSender(applicationInfoPath, clientPort); +if (!applicationInfoPath || !clientPort || !message) { + console.error("Missing necessary environment variables for AnnouncementWorker."); + process.exit(1); +} -// Start the announcement process and handle results -announcementWorker.start(message); +// Initialize the AnnouncementSender instance +const announcementSender = new AnnouncementSender(applicationInfoPath, clientPort); + +// Start the announcement process +announcementSender.start(message); + +// Handle graceful termination on receiving kill signals +const handleExit = async () => { + console.log("Announcement worker is shutting down gracefully..."); + await announcementSender.stop(); // Assuming stop() is implemented to clean up resources + process.exit(0); +}; + +process.on('SIGTERM', handleExit); +process.on('SIGINT', handleExit); diff --git a/CEO/src/workers/servers_worker.ts b/CEO/src/workers/servers_worker.ts index 4e6f4b6..1555807 100644 --- a/CEO/src/workers/servers_worker.ts +++ b/CEO/src/workers/servers_worker.ts @@ -1,14 +1,36 @@ -import {UdpServer} from "../network/udp/udp_server"; -import {TcpServer} from "../network/tcp/tcp_server"; -import {workerData} from "worker_threads"; +import { UdpServer } from "../network/udp/udp_server"; +import { TcpServer } from "../network/tcp/tcp_server"; -let udpServer: UdpServer | null = null; -let tcpServer: TcpServer | null = null; +let udpServer: UdpServer | null +let tcpServer: TcpServer | null -const {USER_UDP_PORT, USER_TCP_PORT, HOST} = workerData; +// Retrieve data from environment variables +const USER_UDP_PORT = parseInt(process.env.USER_UDP_PORT || '0', 10); +const USER_TCP_PORT = parseInt(process.env.USER_TCP_PORT || '0', 10); +const HOST = process.env.HOST || ''; +// Initialize and start the servers udpServer = new UdpServer(HOST, USER_UDP_PORT); udpServer.start(); tcpServer = new TcpServer(HOST, USER_TCP_PORT); -tcpServer.start(); \ No newline at end of file +tcpServer.start(); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + // Perform any cleanup, such as closing connections, saving data, etc. + // Example: if you have a server instance running, you may want to close it: + // await server.close(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/UC/src/network/operations_custom/general_operations.ts b/UC/src/network/operations_custom/general_operations.ts index 4bdee66..f1d84e3 100644 --- a/UC/src/network/operations_custom/general_operations.ts +++ b/UC/src/network/operations_custom/general_operations.ts @@ -7,14 +7,14 @@ export class GeneralOperations implements OperationPlugin { public static readonly operationCodes = { OK: 'OK', ERR: 'ERR', - HEARTBEAT: 'HEARTBEAT', + ARE_YOU_UC: 'ARE_YOU_UC', ALIVE: 'ALIVE', SET_PUBLIC_KEY: 'SET_PUBLIC_KEY', SET_AES_KEY: 'SET_AES_KEY', }; // Handle heartbeat operation asynchronously - public static async handleHeartbeat(): Promise { + public static async handleAreYouUC(): Promise { const networkInterfaces = os.networkInterfaces(); let ipAddress = 'Unknown'; @@ -79,7 +79,7 @@ export class GeneralOperations implements OperationPlugin { // Register general operations with the OperationHandler public register(operationHandler: OperationHandler): void { - operationHandler.registerHandler(GeneralOperations.operationCodes.HEARTBEAT, GeneralOperations.handleHeartbeat); + operationHandler.registerHandler(GeneralOperations.operationCodes.ARE_YOU_UC, GeneralOperations.handleAreYouUC); operationHandler.registerHandler(GeneralOperations.operationCodes.SET_PUBLIC_KEY, GeneralOperations.handlePublicKey); operationHandler.registerHandler(GeneralOperations.operationCodes.SET_AES_KEY, GeneralOperations.handleAESKey); operationHandler.registerHandler(GeneralOperations.operationCodes.OK, GeneralOperations.handleOk); diff --git a/User/src/helpers/backup_manager.ts b/User/src/helpers/backup_manager.ts index 0868cf5..ae87b65 100644 --- a/User/src/helpers/backup_manager.ts +++ b/User/src/helpers/backup_manager.ts @@ -14,6 +14,8 @@ export class BackupManager { private userConfig: JsonManager; private readonly clientPort: number; private isBusy: boolean = false; + private intervalId: NodeJS.Timeout | null = null; + private stopRequested: boolean = false; constructor(userConfigPath: string, applicationInfoPath: string, memoryManagerPath: string, clientPort: number) { this.applicationInfo = new JsonManager(applicationInfoPath); @@ -23,8 +25,8 @@ export class BackupManager { } async start(): Promise { - setInterval(async () => { - if (!this.isBusy) { + this.intervalId = setInterval(async () => { + if (!this.isBusy || !this.stopRequested) { this.isBusy = true; this.log('Start successfully. Backup files to users.'); await this.initialize(); @@ -160,6 +162,22 @@ export class BackupManager { }); } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + // Unified logging function private log(message: string, level: 'log' | 'error' | 'warn' = 'log'): void { const prefix = '[BackupManager]'; diff --git a/User/src/helpers/backup_retrieval.ts b/User/src/helpers/backup_retrieval.ts index 43ef3d1..aec9696 100644 --- a/User/src/helpers/backup_retrieval.ts +++ b/User/src/helpers/backup_retrieval.ts @@ -14,6 +14,8 @@ export class BackupRetrievalWorker { private encryptionKey: Buffer | null = null; private iv: Buffer | null = null; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; + private isBusy: boolean = false; constructor(userConfigPath: string, applicationInfoPath: string, clientPort: number, destinationPath: string) { this.userConfig = new JsonManager(userConfigPath); @@ -33,6 +35,8 @@ export class BackupRetrievalWorker { } async start(): Promise { + if(!this.stopRequested) return; + this.isBusy = true; try { const userInfo = await this.userConfig.readValue('user_info'); if (!userInfo || !userInfo.name) { @@ -61,10 +65,15 @@ export class BackupRetrievalWorker { this.log(`Backup retrieved successfully from ${ip}`); } - process.send?.({ type: 'log', message: 'Backup retrieval completed successfully.' }); + process.send?.({ type: 'showAlert', message: 'Backup retrieval completed successfully.' }); + process.send?.({ type: 'changeContent', page: 'main_menu' }); } catch (error: any) { this.log(`Error in BackupRetrievalWorker: ${error.message}`, 'error'); - process.send?.({ type: 'log', message: `A problem occurred: ${error.message}` }); + process.send?.({ type: 'showAlert', message: `A problem occurred: ${error.message}` }); + process.send?.({ type: 'changeContent', page: 'main_menu' }); + } + finally { + this.isBusy = false; } if (global.gc) { @@ -172,6 +181,17 @@ export class BackupRetrievalWorker { } } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { diff --git a/User/src/helpers/department_sharer.ts b/User/src/helpers/department_sharer.ts index e084811..ea844b7 100644 --- a/User/src/helpers/department_sharer.ts +++ b/User/src/helpers/department_sharer.ts @@ -14,6 +14,8 @@ export class DepartmentSharer { private readonly clientPort: number; private isBusy: boolean = false; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; + private intervalId: NodeJS.Timeout | null = null; constructor( userConfigPath: string, @@ -30,8 +32,8 @@ export class DepartmentSharer { // Start sharing files with the department every minute async start(): Promise { - setInterval(async () => { - if (!this.isBusy) { + this.intervalId = setInterval(async () => { + if (!this.isBusy || !this.stopRequested) { this.isBusy = true; this.log('Start successfully. Sharing files with the department.'); await this.shareFilesWithDepartment(); @@ -172,6 +174,22 @@ export class DepartmentSharer { } } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { diff --git a/User/src/helpers/file_sharer.ts b/User/src/helpers/file_sharer.ts index 72ad245..cada88f 100644 --- a/User/src/helpers/file_sharer.ts +++ b/User/src/helpers/file_sharer.ts @@ -17,6 +17,8 @@ export class FileSharer { private readonly clientPort: number; private isBusy: boolean; private tcpCommunicator: TcpCommunicator | null = null; + private stopRequested: boolean = false; + private intervalId: NodeJS.Timeout | null = null; constructor(queueFilePath: string, clientPort: number) { this.queueManager = new QueueManager(queueFilePath, compareFnFileItemTask); @@ -26,8 +28,8 @@ export class FileSharer { // Start processing the file queue async start(): Promise { - setInterval(async () => { - if (!this.isBusy) { // Check if the queue is already being processed + this.intervalId = setInterval(async () => { + if (!this.isBusy || !this.stopRequested) { // Check if the queue is already being processed this.isBusy = true; // Set busy flag to true before starting this.log("Start successfully. Processing the queue."); await this.processQueue(); @@ -104,6 +106,22 @@ export class FileSharer { return true; } + async stop(): Promise { + this.stopRequested = true; // Signal that stop is requested + + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + } + + // Wait for any ongoing process to complete if busy + while (this.isBusy) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + console.log("[BackupManager] Stopped successfully."); + } + private async waitForResponse(): Promise { return new Promise((resolve) => { const idResponseCheck = setInterval(async () => { diff --git a/User/src/helpers/network_scanner.ts b/User/src/helpers/network_scanner.ts index 926f274..d1ce897 100644 --- a/User/src/helpers/network_scanner.ts +++ b/User/src/helpers/network_scanner.ts @@ -54,7 +54,7 @@ export class NetworkScanner { try { this.log('UC Check running...', 'log', 'startUCCheck'); const udpClient = new UdpClient(this.udpPort); - const aliveClients = await udpClient.getAliveClients(); + const aliveClients = await udpClient.getTargetClients(operationCodes.ARE_YOU_UC); const storedIp = await this.applicationInfo.readValue('serverIp'); const foundClient = aliveClients.length > 0; @@ -95,7 +95,7 @@ export class NetworkScanner { this.log('IP Lookup running...', 'log', 'startUserIPLookup'); const serverIp = await this.applicationInfo.readValue('serverIp'); const udpClient = new UdpClient(this.udpPort); - const activeIPs = await udpClient.getAliveClients(); + const activeIPs = await udpClient.getTargetClients(operationCodes.ARE_YOU_HUMAN); const filteredIPs = activeIPs.filter(ip => ip !== serverIp); // Save the filtered IPs to 'users_ip' diff --git a/User/src/helpers/users_info_fetcher.ts b/User/src/helpers/users_info_fetcher.ts index fae95f5..2852b04 100644 --- a/User/src/helpers/users_info_fetcher.ts +++ b/User/src/helpers/users_info_fetcher.ts @@ -11,6 +11,7 @@ export class UsersInfoFetcher { private readonly clientPort: number; private memoryId: string; private readonly activeUsersKey: string; + private intervalId: NodeJS.Timeout | null = null; constructor(applicationInfoPath: string, memoryManagerPath: string, clientPort: number) { this.applicationInfo = new JsonManager(applicationInfoPath); @@ -23,13 +24,13 @@ export class UsersInfoFetcher { // Method to start checking user info periodically (every minute) async start(): Promise { - setInterval(async () => { + this.intervalId = setInterval(async () => { await this.initialize(); // Re-run every minute if (global.gc) { global.gc(); } - }, 5000); // 1 minute interval + }, 5000); // 5-second interval for testing } // Initialize and fetch user IPs and process users info @@ -101,6 +102,14 @@ export class UsersInfoFetcher { await this.memoryManager.updateMetaInformation(this.memoryId, userInfo); // Update active users in memory } + stop(): void { + if (this.intervalId) { + clearInterval(this.intervalId); + this.intervalId = null; + console.log("[UsersInfoFetcher] Stopped successfully."); + } + } + // Unified logging function private log(message: string, level: 'log' | 'error' = 'log'): void { const prefix = '[UsersInfoFetcher]'; diff --git a/User/src/helpers/worker_manager.ts b/User/src/helpers/worker_manager.ts index 3f5e066..c4980b6 100644 --- a/User/src/helpers/worker_manager.ts +++ b/User/src/helpers/worker_manager.ts @@ -5,7 +5,8 @@ import { WindowManager } from "./window_manager"; export class WorkerManager { private readonly pathToWorkerDir: string; private windowManager: WindowManager; - private workers: ChildProcess[]; // Array to store running child processes + private workers: ChildProcess[]; + private cleanupInProgress: boolean = false; constructor(pathToWorkerDir: string, windowManager: WindowManager) { this.pathToWorkerDir = pathToWorkerDir; @@ -87,17 +88,24 @@ export class WorkerManager { reject(err); }); - worker.on('exit', (code) => { - console.log(`${scriptName} exited with code ${code}`); + worker.on('exit', (code, signal) => { this.removeWorker(worker); - if (code === 0) resolve(); - else reject(new Error(`${scriptName} exited with code ${code}`)); + if (code === 0) { + console.log(`${scriptName} exited successfully`); + resolve(); + } else if (signal) { + console.log(`${scriptName} was killed with signal: ${signal}`); + } else { + console.error(`${scriptName} exited with code: ${code}`);; + } }); }); } - // Close all running workers closeAllWorkers(): void { + if (this.cleanupInProgress) return; // Prevent duplicate cleanup + this.cleanupInProgress = true; + console.log('Terminating all running workers...'); this.workers.forEach(worker => worker.kill()); this.workers = []; diff --git a/User/src/network/operation_codes.ts b/User/src/network/operation_codes.ts index 7f6edb1..fe7bd78 100644 --- a/User/src/network/operation_codes.ts +++ b/User/src/network/operation_codes.ts @@ -1,6 +1,7 @@ export let operationCodes = { // General Operations - HEARTBEAT: 'HEARTBEAT', + ARE_YOU_HUMAN: 'ARE_YOU_HUMAN', + ARE_YOU_UC: 'ARE_YOU_UC', ALIVE: 'ALIVE', SET_PUBLIC_KEY: 'SET_PUBLIC_KEY', SET_AES_KEY: 'SET_AES_KEY', diff --git a/User/src/network/operations_custom/general_operations.ts b/User/src/network/operations_custom/general_operations.ts index 4bdee66..afe4c50 100644 --- a/User/src/network/operations_custom/general_operations.ts +++ b/User/src/network/operations_custom/general_operations.ts @@ -7,14 +7,14 @@ export class GeneralOperations implements OperationPlugin { public static readonly operationCodes = { OK: 'OK', ERR: 'ERR', - HEARTBEAT: 'HEARTBEAT', + ARE_YOU_HUMAN: 'ARE_YOU_HUMAN', ALIVE: 'ALIVE', SET_PUBLIC_KEY: 'SET_PUBLIC_KEY', SET_AES_KEY: 'SET_AES_KEY', }; // Handle heartbeat operation asynchronously - public static async handleHeartbeat(): Promise { + public static async handleAreYouHuman(): Promise { const networkInterfaces = os.networkInterfaces(); let ipAddress = 'Unknown'; @@ -79,7 +79,7 @@ export class GeneralOperations implements OperationPlugin { // Register general operations with the OperationHandler public register(operationHandler: OperationHandler): void { - operationHandler.registerHandler(GeneralOperations.operationCodes.HEARTBEAT, GeneralOperations.handleHeartbeat); + operationHandler.registerHandler(GeneralOperations.operationCodes.ARE_YOU_HUMAN, GeneralOperations.handleAreYouHuman); operationHandler.registerHandler(GeneralOperations.operationCodes.SET_PUBLIC_KEY, GeneralOperations.handlePublicKey); operationHandler.registerHandler(GeneralOperations.operationCodes.SET_AES_KEY, GeneralOperations.handleAESKey); operationHandler.registerHandler(GeneralOperations.operationCodes.OK, GeneralOperations.handleOk); diff --git a/User/src/network/udp/udp_client.ts b/User/src/network/udp/udp_client.ts index 2449639..a16c2e1 100644 --- a/User/src/network/udp/udp_client.ts +++ b/User/src/network/udp/udp_client.ts @@ -29,7 +29,7 @@ export class UdpClient { } // Send heartbeat and get clients that respond with ALIVE, excluding the host machine's IP addresses - async getAliveClients(): Promise { + async getTargetClients(heartbeatCode: string): Promise { const subnet = this.getSubnet(); const ipRange = this.getIPRange(subnet); @@ -45,7 +45,7 @@ export class UdpClient { const aliveClients: string[] = []; for (const ip of activeIps) { if (!localIPs.includes(ip)) { - const result = await this.sendHeartbeat(ip); + const result = await this.sendHeartbeat(ip, heartbeatCode); if (result.found) { aliveClients.push(ip); } @@ -73,9 +73,8 @@ export class UdpClient { } // Send heartbeat to an IP - private async sendHeartbeat(ip: string): Promise<{ found: boolean }> { + private async sendHeartbeat(ip: string, heartbeatCode: string): Promise<{ found: boolean }> { return new Promise((resolve) => { - const heartbeatCode = operationCodes.HEARTBEAT; const heartbeatMessage = MessageHandler.formatMessage(heartbeatCode); this.log(`Sending heartbeat to ${ip}`); diff --git a/User/src/workers/backup_retrieval_worker.ts b/User/src/workers/backup_retrieval_worker.ts index fa58038..ef1a3dd 100644 --- a/User/src/workers/backup_retrieval_worker.ts +++ b/User/src/workers/backup_retrieval_worker.ts @@ -25,4 +25,26 @@ const backupRetrievalWorker = new BackupRetrievalWorker( ); // Start the backup retrieval process -backupRetrievalWorker.start(); +backupRetrievalWorker.start().then(() => { + console.log('Backup retrieval process completed successfully.'); +}); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + // Perform any cleanup, such as closing connections, saving data, etc. + // Example: if you have a server instance running, you may want to close it: + // await server.close(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} + diff --git a/User/src/workers/directories_watcher_worker.ts b/User/src/workers/directories_watcher_worker.ts index 288dd5c..b864f40 100644 --- a/User/src/workers/directories_watcher_worker.ts +++ b/User/src/workers/directories_watcher_worker.ts @@ -23,3 +23,22 @@ departmentShareManager.start(); const shareFileManager = new DirectoryWatcher(memoryManagerPath, applicationInfoPath, 'shareDirectory'); shareFileManager.start(); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + backupDirectoryManager.closeWatcher(); + departmentShareManager.closeWatcher(); + shareFileManager.closeWatcher(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/User/src/workers/network_scanner_worker.ts b/User/src/workers/network_scanner_worker.ts index ee86577..f60bf03 100644 --- a/User/src/workers/network_scanner_worker.ts +++ b/User/src/workers/network_scanner_worker.ts @@ -20,3 +20,20 @@ const networkScanner = new NetworkScanner( errorPage, databaseResetPage ); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + networkScanner.stopAllIntervals(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/User/src/workers/resource_coordinator_worker.ts b/User/src/workers/resource_coordinator_worker.ts index 036d79e..c10513c 100644 --- a/User/src/workers/resource_coordinator_worker.ts +++ b/User/src/workers/resource_coordinator_worker.ts @@ -21,3 +21,23 @@ fileSharer.start(); const departmentSharer = new DepartmentSharer(usersConfigPath, applicationInfoPath, memoryManagerPath, tcpPort); departmentSharer.start(); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + usersInfoFetcher.stop(); + await backupManager.stop(); + await fileSharer.stop(); + await departmentSharer.stop(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +} diff --git a/User/src/workers/servers_worker.ts b/User/src/workers/servers_worker.ts index 40bb6e7..1555807 100644 --- a/User/src/workers/servers_worker.ts +++ b/User/src/workers/servers_worker.ts @@ -1,8 +1,8 @@ import { UdpServer } from "../network/udp/udp_server"; import { TcpServer } from "../network/tcp/tcp_server"; -let udpServer: UdpServer | null = null; -let tcpServer: TcpServer | null = null; +let udpServer: UdpServer | null +let tcpServer: TcpServer | null // Retrieve data from environment variables const USER_UDP_PORT = parseInt(process.env.USER_UDP_PORT || '0', 10); @@ -15,3 +15,22 @@ udpServer.start(); tcpServer = new TcpServer(HOST, USER_TCP_PORT); tcpServer.start(); + +process.on('SIGTERM', async () => { + console.log('Received SIGTERM. Cleaning up...'); + await cleanupAndExit(); +}); + +process.on('SIGINT', async () => { + console.log('Received SIGINT. Cleaning up...'); + await cleanupAndExit(); +}); + +async function cleanupAndExit() { + // Perform any cleanup, such as closing connections, saving data, etc. + // Example: if you have a server instance running, you may want to close it: + // await server.close(); + + console.log('Cleanup complete. Exiting.'); + process.exit(0); // Exit with code 0 to indicate a clean exit +}