BACKEND DONE FOR ALL APPS
This commit is contained in:
andrei-mihnea-cerbu
@@ -0,0 +1,205 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { FileEncryptor } from './file_encryptor'; // Assume the class is in this file
|
||||
import { MemoryManager } from './memory_manager'; // Assume this handles memory-based storage
|
||||
import { JsonManager } from './json_manager'; // Manages JSON-based configurations
|
||||
import { TcpClient } from '../network/tcp/tcp_client' // Import your TcpClient class
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import { parentPort } from 'worker_threads';
|
||||
|
||||
export class BackupManager {
|
||||
private fileEncryptor: FileEncryptor | null = null;
|
||||
private memoryManager: MemoryManager;
|
||||
private applicationInfo: JsonManager;
|
||||
private userConfig: JsonManager;
|
||||
private readonly clientPort: number;
|
||||
|
||||
constructor(userConfigPath: string, applicationInfoPath: string, memoryManagerPath: string, clientPort: number) {
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.userConfig = new JsonManager(userConfigPath);
|
||||
this.memoryManager = new MemoryManager(memoryManagerPath);
|
||||
this.clientPort = clientPort;
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
await this.initialize(); // Re-run every minute
|
||||
}, 60000); // 1 minute interval
|
||||
}
|
||||
|
||||
// Initialize the backup process: fetch data from the app info and memory
|
||||
private async initialize(): Promise<void> {
|
||||
// Initialize the encryption settings from UserConfig
|
||||
const encryptionKeyData = await this.userConfig.readValue('encryption_key');
|
||||
if (!encryptionKeyData || !encryptionKeyData.key || !encryptionKeyData.iv) {
|
||||
console.error('Encryption key data is missing in user configuration.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Create a new FileEncryptor with the retrieved key and IV
|
||||
this.fileEncryptor = new FileEncryptor(encryptionKeyData.key, encryptionKeyData.iv);
|
||||
|
||||
// Get the name of the user
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.name) {
|
||||
console.error('User information is missing in user configuration.');
|
||||
return;
|
||||
}
|
||||
|
||||
const userName = userInfo.name;
|
||||
|
||||
// Get the backup directory information from ApplicationInfo
|
||||
const backupDirectoryData = await this.applicationInfo.readValue('backupDirectory');
|
||||
if (!backupDirectoryData || !backupDirectoryData.id || !backupDirectoryData.path) {
|
||||
console.error('Backup directory information is missing in application info.');
|
||||
return;
|
||||
}
|
||||
|
||||
const activeUsersIp = await this.applicationInfo.readValue('users_ip');
|
||||
if (!activeUsersIp || !activeUsersIp.length) {
|
||||
console.error('No active users found.');
|
||||
return;
|
||||
}
|
||||
|
||||
const backupDirectoryId = backupDirectoryData.id;
|
||||
const backupDirectoryPath = backupDirectoryData.path;
|
||||
|
||||
// Get the file structure from MemoryManager
|
||||
const directoryData = await this.memoryManager.retrieveMetaInformation(backupDirectoryId);
|
||||
if (!directoryData || !directoryData.structure) {
|
||||
console.error('Backup directory structure is missing in memory.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Send files to the list of IPs
|
||||
await this.sendFilesToUsers(directoryData.structure, userName, activeUsersIp, backupDirectoryPath);
|
||||
}
|
||||
|
||||
// Method to encrypt a file and return the base64 string
|
||||
private encryptFile(filePath: string): string {
|
||||
if (!this.fileEncryptor) {
|
||||
return filePath;
|
||||
}
|
||||
|
||||
if (!fs.existsSync(filePath)) {
|
||||
console.error(`File not found: ${filePath}`);
|
||||
return '';
|
||||
}
|
||||
|
||||
// Use FileEncryptor to encrypt the file and return the base64 string
|
||||
return this.fileEncryptor.encryptFileToBase64(filePath);
|
||||
}
|
||||
|
||||
// Send files to the list of users and remove successfully sent files from the list
|
||||
private async sendFilesToUsers(fileStructure: { [key: string]: string }, userName: string, usersIp: string[], backupDirectoryPath: string): Promise<void> {
|
||||
let unsentFiles = Object.keys(fileStructure); // Keep track of unsent files
|
||||
|
||||
for (const fileName of unsentFiles) {
|
||||
const filePath = fileStructure[fileName];
|
||||
const encryptedFileContent = this.encryptFile(filePath);
|
||||
|
||||
if (!encryptedFileContent) {
|
||||
console.error(`Failed to encrypt file: ${fileName}`);
|
||||
continue; // Skip to the next file
|
||||
}
|
||||
|
||||
// Calculate relative file path
|
||||
const relativeFilePath = path.relative(backupDirectoryPath, filePath); // Get the relative file path
|
||||
|
||||
// Meta information to send
|
||||
const metaInfo = {
|
||||
userName, // Name of the user
|
||||
relativeFilePath // Relative path to preserve the directory structure
|
||||
};
|
||||
|
||||
for (const ip of usersIp) {
|
||||
const tcpClient = new TcpClient(this.clientPort);
|
||||
tcpClient.openSocket(ip);
|
||||
|
||||
// Wait for AES key and send the file
|
||||
try {
|
||||
const sendSuccess = await this.waitForAesKeyAndSendFile(tcpClient, operationCodes.BACKUP_FILE, metaInfo, Buffer.from(encryptedFileContent, 'base64'));
|
||||
if (sendSuccess) {
|
||||
console.log(`Successfully sent file: ${fileName} to ${ip}`);
|
||||
unsentFiles = unsentFiles.filter(f => f !== fileName); // Remove the file from the unsent list
|
||||
tcpClient.closeSocket();
|
||||
break; // Move to the next file after successful send
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Failed to send file: ${fileName} to ${ip}. Error: ${error}`);
|
||||
tcpClient.closeSocket();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If there are any unsent files, notify the parent process
|
||||
if (unsentFiles.length > 0) {
|
||||
parentPort?.postMessage({ success: false, message: 'Backup could not be completed for all files', unsentFiles });
|
||||
} else {
|
||||
parentPort?.postMessage({ success: true, message: 'Backup completed successfully' });
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for AES key to be set, send the file, and wait for the response
|
||||
private async waitForAesKeyAndSendFile(tcpClient: TcpClient, operationCode: string, metaInfo: { [key: string]: any }, fileContent: Buffer, checkInterval: number = 500, timeout: number = 10000): Promise<boolean> {
|
||||
const startTime = Date.now();
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const intervalId = setInterval(async () => {
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
// Check if AES key is set, if timeout occurs, reject
|
||||
if (!tcpClient.isAesKeySet()) {
|
||||
if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
console.error('Timeout waiting for AES key.');
|
||||
reject(new Error('Timeout waiting for AES key.'));
|
||||
}
|
||||
return; // Continue waiting for AES key
|
||||
}
|
||||
|
||||
// AES key is set, send the message
|
||||
clearInterval(intervalId);
|
||||
|
||||
try {
|
||||
const success = await tcpClient.sendMessage(operationCode, metaInfo, fileContent);
|
||||
if (!success) {
|
||||
reject(new Error('Failed to send the file content.'));
|
||||
return;
|
||||
}
|
||||
|
||||
// Wait for a response after sending the message
|
||||
const responseReceived = await this.waitForMessageResponse(tcpClient, timeout);
|
||||
if (!responseReceived) {
|
||||
reject(new Error('Timeout waiting for the message response.'));
|
||||
return;
|
||||
}
|
||||
|
||||
// Everything went fine
|
||||
resolve(true);
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
}, checkInterval); // Check for AES key every `checkInterval`
|
||||
});
|
||||
}
|
||||
|
||||
// Method to wait for the response from the TCP client
|
||||
private waitForMessageResponse(tcpClient: TcpClient, timeout: number = 10000): Promise<any> {
|
||||
return new Promise((resolve) => {
|
||||
const startTime = Date.now();
|
||||
const intervalId = setInterval(() => {
|
||||
const response = tcpClient.getLastResult();
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
if (response) {
|
||||
clearInterval(intervalId);
|
||||
resolve(true);
|
||||
} else if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
resolve(false); // No response after timeout
|
||||
}
|
||||
}, 100);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,224 @@
|
||||
import { JsonManager } from './json_manager'; // Assuming this manages JSON configurations
|
||||
import { TcpClient } from '../network/tcp/tcp_client'; // Assuming this handles TCP client connections
|
||||
import { operationCodes } from '../network/operation_codes'; // Assuming this holds operation codes
|
||||
import { parentPort } from 'worker_threads';
|
||||
import path from 'path';
|
||||
import fs from 'fs';
|
||||
import crypto from 'crypto'; // Import the crypto module for encryption and decryption
|
||||
|
||||
export class BackupRetrievalWorker {
|
||||
private userConfig: JsonManager;
|
||||
private applicationInfo: JsonManager;
|
||||
private readonly clientPort: number;
|
||||
private readonly destinationPath: string;
|
||||
private encryptionKey: Buffer | null = null;
|
||||
private iv: Buffer | null = null;
|
||||
|
||||
constructor(userConfigPath: string, applicationInfoPath: string, clientPort: number, destinationPath: string) {
|
||||
this.userConfig = new JsonManager(userConfigPath);
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.clientPort = clientPort;
|
||||
this.destinationPath = destinationPath;
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
// Retrieve the necessary data from userConfig and applicationInfo
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.name) {
|
||||
console.error('User information or name is missing.');
|
||||
return;
|
||||
}
|
||||
const userName = userInfo.name;
|
||||
|
||||
// Load encryption key and IV from userConfig
|
||||
const encryptionData = await this.userConfig.readValue('encryption_key');
|
||||
if (!encryptionData || !encryptionData.key || !encryptionData.iv) {
|
||||
console.error('Encryption key or IV is missing.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Convert encryption key and IV from base64 to buffer
|
||||
this.encryptionKey = Buffer.from(encryptionData.key, 'base64');
|
||||
this.iv = Buffer.from(encryptionData.iv, 'base64');
|
||||
|
||||
const activeUsersIp = await this.applicationInfo.readValue('users_ip');
|
||||
if (!activeUsersIp || !activeUsersIp.length) {
|
||||
console.error('No active users found.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Process each IP and request backup data
|
||||
for (const ip of activeUsersIp) {
|
||||
try {
|
||||
const success = await this.processBackupForIp(ip, userName);
|
||||
if (success) {
|
||||
console.log(`Backup retrieved successfully from ${ip}`);
|
||||
} else {
|
||||
console.error(`Failed to retrieve backup from ${ip}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Error processing backup from ${ip}: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Notify the parent that the worker is done
|
||||
parentPort?.postMessage({ success: true, message: 'Backup retrieval completed.' });
|
||||
}
|
||||
|
||||
private async processBackupForIp(ip: string, userName: string): Promise<boolean> {
|
||||
const tcpClient = new TcpClient(this.clientPort);
|
||||
tcpClient.openSocket(ip);
|
||||
|
||||
try {
|
||||
// Wait for the AES key to be set before continuing
|
||||
const aesSet = await this.waitForAesKey(tcpClient);
|
||||
if (!aesSet) {
|
||||
console.error(`Timeout waiting for AES key on IP ${ip}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Step 1: Check if a backup exists for the user on this IP
|
||||
const backupExists = await this.checkIfBackupExists(tcpClient, userName);
|
||||
if (!backupExists) {
|
||||
console.log(`No backup found for user ${userName} on IP ${ip}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Step 2: Request the structure of the backup directory
|
||||
const backupStructure = await this.requestBackupStructure(tcpClient, userName);
|
||||
if (!backupStructure || Object.keys(backupStructure).length === 0) {
|
||||
console.log(`No files found in backup structure for user ${userName} on IP ${ip}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Step 3: Request and retrieve each file from the backup
|
||||
for (const relativeFilePath of Object.keys(backupStructure)) {
|
||||
const fileRequestSuccess = await this.requestBackupFile(tcpClient, userName, relativeFilePath);
|
||||
if (!fileRequestSuccess) {
|
||||
console.error(`Failed to retrieve file ${relativeFilePath} from backup for user ${userName} on IP ${ip}`);
|
||||
return false; // Stop if any file fails to be retrieved
|
||||
}
|
||||
}
|
||||
|
||||
return true; // All files retrieved successfully
|
||||
} catch (error) {
|
||||
console.error(`Error during backup processing for IP ${ip}: ${error}`);
|
||||
return false;
|
||||
} finally {
|
||||
tcpClient.closeSocket(); // Ensure socket is closed
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for AES key to be set
|
||||
private async waitForAesKey(tcpClient: TcpClient, timeout: number = 10000): Promise<boolean> {
|
||||
const startTime = Date.now();
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const intervalId = setInterval(() => {
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
if (tcpClient.isAesKeySet()) {
|
||||
clearInterval(intervalId);
|
||||
resolve(true); // AES key is set, we can proceed
|
||||
} else if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
resolve(false); // Timeout reached, AES key not set
|
||||
}
|
||||
}, 500); // Check every 500ms
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for a message response with timeout
|
||||
private async waitForMessageResponse(tcpClient: TcpClient, timeout: number = 10000): Promise<any> {
|
||||
const startTime = Date.now();
|
||||
|
||||
return new Promise((resolve) => {
|
||||
const intervalId = setInterval(() => {
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
const response = tcpClient.getLastResult();
|
||||
|
||||
if (response) {
|
||||
clearInterval(intervalId);
|
||||
resolve(response); // Return the response
|
||||
} else if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
resolve(null); // Timeout, no response
|
||||
}
|
||||
}, 500); // Check every 500ms
|
||||
});
|
||||
}
|
||||
|
||||
// Check if the backup exists for the user on the remote IP
|
||||
private async checkIfBackupExists(tcpClient: TcpClient, userName: string): Promise<boolean> {
|
||||
const metaInfo = { name: userName };
|
||||
await tcpClient.sendMessage(operationCodes.IS_BACKUP_CREATED, metaInfo);
|
||||
|
||||
const response = await this.waitForMessageResponse(tcpClient);
|
||||
return response?.metaInfo?.backupExists === true;
|
||||
}
|
||||
|
||||
// Request the backup structure from the remote IP
|
||||
private async requestBackupStructure(tcpClient: TcpClient, userName: string): Promise<any> {
|
||||
const metaInfo = { name: userName };
|
||||
await tcpClient.sendMessage(operationCodes.GET_BACKUP_STRUCTURE, metaInfo);
|
||||
|
||||
const response = await this.waitForMessageResponse(tcpClient);
|
||||
return response?.metaInfo?.structure ? response.metaInfo.structure : null;
|
||||
}
|
||||
|
||||
// Request a file from the backup and wait for it to be decrypted and stored
|
||||
private async requestBackupFile(tcpClient: TcpClient, userName: string, relativeFilePath: string): Promise<boolean> {
|
||||
const metaInfo = { name: userName, relativeFilePath };
|
||||
await tcpClient.sendMessage(operationCodes.REQ_FILE_FROM_BACKUP, metaInfo);
|
||||
|
||||
const response = await this.waitForMessageResponse(tcpClient);
|
||||
if (response?.operationCode === operationCodes.OK && response.metaInfo && response.fileContent) {
|
||||
return this.saveFile(response.metaInfo.relativeFilePath, response.fileContent);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
// Decrypt the file content using AES-256-CBC and save the decrypted file
|
||||
private saveFile(relativeFilePath: string, fileContent: string): boolean {
|
||||
if (!this.encryptionKey || !this.iv) {
|
||||
console.error('Encryption key or IV is not set.');
|
||||
return false;
|
||||
}
|
||||
|
||||
// Decode the base64-encoded file content into a buffer
|
||||
let encryptedBuffer: Buffer;
|
||||
try {
|
||||
encryptedBuffer = Buffer.from(fileContent, 'base64');
|
||||
} catch (error) {
|
||||
console.error('Error decoding base64 file content:', error);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Decrypt the file content
|
||||
let decryptedContent: Buffer;
|
||||
try {
|
||||
const decipher = crypto.createDecipheriv('aes-256-cbc', this.encryptionKey, this.iv);
|
||||
decryptedContent = Buffer.concat([decipher.update(encryptedBuffer), decipher.final()]);
|
||||
} catch (error) {
|
||||
console.error('Error decrypting file:', error);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Save the decrypted file content
|
||||
const fullFilePath = path.join(this.destinationPath, relativeFilePath);
|
||||
|
||||
try {
|
||||
const dirPath = path.dirname(fullFilePath);
|
||||
if (!fs.existsSync(dirPath)) {
|
||||
fs.mkdirSync(dirPath, { recursive: true }); // Ensure directory exists
|
||||
}
|
||||
|
||||
fs.writeFileSync(fullFilePath, decryptedContent);
|
||||
console.log(`File saved successfully: ${fullFilePath}`);
|
||||
return true;
|
||||
} catch (error: any) {
|
||||
console.error(`Error saving file ${relativeFilePath}: ${error.message}`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,278 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { TcpClient } from '../network/tcp/tcp_client'; // Assuming this class exists
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import { JsonManager } from './json_manager'; // Manages JSON configurations
|
||||
import { MemoryManager } from './memory_manager'; // Manages in-memory data structures
|
||||
|
||||
export class DepartmentSharer {
|
||||
private userConfig: JsonManager;
|
||||
private applicationInfo: JsonManager;
|
||||
private memoryManager: MemoryManager; // To read the department files
|
||||
private departmentDirectory: string | null;
|
||||
private readonly clientPort: number;
|
||||
|
||||
constructor(
|
||||
userConfigPath: string,
|
||||
applicationInfoPath: string,
|
||||
memoryManagerPath: string,
|
||||
clientPort: number
|
||||
) {
|
||||
this.userConfig = new JsonManager(userConfigPath);
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.memoryManager = new MemoryManager(memoryManagerPath); // To retrieve files
|
||||
this.clientPort = clientPort;
|
||||
this.departmentDirectory = null;
|
||||
}
|
||||
|
||||
// Start sharing files with the department every minute
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
await this.shareFilesWithDepartment(); // Retry every minute
|
||||
}, 10000); // 10-second interval for testing
|
||||
}
|
||||
|
||||
// Share files with users in the same department
|
||||
private async shareFilesWithDepartment(): Promise<void> {
|
||||
console.log('\n\nStarting Department Share Process\n\n');
|
||||
|
||||
// Get the current user's department information
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.departmentId || !userInfo.name) {
|
||||
console.error('User information or department ID is missing in the configuration.');
|
||||
return;
|
||||
}
|
||||
|
||||
const departmentId = userInfo.departmentId;
|
||||
const userName = userInfo.name;
|
||||
|
||||
// Get the list of active users from applicationInfo
|
||||
const activeUsersId = await this.applicationInfo.readValue('active_users_info');
|
||||
if (!activeUsersId) {
|
||||
console.error('No active users found.');
|
||||
return;
|
||||
}
|
||||
|
||||
const activeUsers = await this.memoryManager.retrieveMetaInformation(activeUsersId);
|
||||
if (!activeUsers || activeUsers.length === 0) {
|
||||
console.error('No active users found.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Filter users who belong to the same department
|
||||
const departmentUsers = activeUsers.filter((user: any) => user.user_info.departmentId === departmentId);
|
||||
|
||||
if (departmentUsers.length === 0) {
|
||||
console.log('No users found in the same department.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Get department directory info
|
||||
const departmentData = await this.applicationInfo.readValue('departmentDirectory');
|
||||
if (!departmentData || !departmentData.path || !departmentData.id) {
|
||||
console.error('No department directory found.');
|
||||
return;
|
||||
}
|
||||
|
||||
this.departmentDirectory = departmentData.path;
|
||||
|
||||
// Read files from the MemoryManager related to this department
|
||||
const departmentFiles = await this.memoryManager.retrieveMetaInformation(departmentData.id);
|
||||
if (!departmentFiles || !departmentFiles.structure) {
|
||||
console.error('No files found for this department in the memory manager.');
|
||||
return;
|
||||
}
|
||||
|
||||
// Send each file to every department user
|
||||
await this.shareFilesWithUsers(departmentUsers, departmentFiles.structure, userName);
|
||||
|
||||
// After sending all files, clear the department directory
|
||||
await this.clearDepartmentDirectory(departmentUsers);
|
||||
}
|
||||
|
||||
// Send the files to the users in the department
|
||||
private async shareFilesWithUsers(users: any[], files: { [key: string]: string }, userName: string): Promise<void> {
|
||||
const unsentFiles = Object.keys(files); // Keep track of unsent files
|
||||
|
||||
for (const fileName of unsentFiles) {
|
||||
const filePath = files[fileName];
|
||||
|
||||
// Ensure the file exists before attempting to send
|
||||
if (!fs.existsSync(filePath)) {
|
||||
console.error(`File not found: ${filePath}`);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Read the file content
|
||||
const fileContent = fs.readFileSync(filePath);
|
||||
|
||||
// Get the relative path of the file (used in the meta info)
|
||||
if (!this.departmentDirectory) return;
|
||||
const relativeFilePath = path.relative(this.departmentDirectory, filePath);
|
||||
|
||||
// Prepare the metaInfo (same structure as FileSharer)
|
||||
const metaInfo = {
|
||||
userName, // Sender's username
|
||||
relativeFilePath // Use the relative path to preserve directory structure
|
||||
};
|
||||
|
||||
// Send the file to each user in the same department
|
||||
for (const user of users) {
|
||||
const userIp = user.ip;
|
||||
const tcpClient = new TcpClient(this.clientPort);
|
||||
|
||||
// Open a connection to the user's IP
|
||||
tcpClient.openSocket(userIp);
|
||||
|
||||
try {
|
||||
// Wait for the AES key and send the file
|
||||
const success = await this.waitForAesKeyAndSendFile(tcpClient, operationCodes.DEPARTMENT_FILE, metaInfo, fileContent);
|
||||
if (success) {
|
||||
console.log(`File successfully sent: ${fileName} to user ${userIp}`);
|
||||
unsentFiles.splice(unsentFiles.indexOf(fileName), 1); // Remove successfully sent file
|
||||
tcpClient.closeSocket(); // Close the connection after sending
|
||||
break; // Move to the next file after a successful send
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Failed to send file: ${fileName} to IP: ${userIp}. Error: ${error}`);
|
||||
tcpClient.closeSocket(); // Ensure socket is closed on error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (unsentFiles.length > 0) {
|
||||
console.log('Some files could not be sent, retrying later.');
|
||||
} else {
|
||||
console.log('All files shared successfully.');
|
||||
}
|
||||
}
|
||||
|
||||
// Clear the department directory for all users after sharing
|
||||
private async clearDepartmentDirectory(users: any[]): Promise<void> {
|
||||
for (const user of users) {
|
||||
const userIp = user.ip;
|
||||
const tcpClient = new TcpClient(this.clientPort);
|
||||
|
||||
// Open a connection to the user's IP
|
||||
tcpClient.openSocket(userIp);
|
||||
|
||||
try {
|
||||
const success = await this.waitForAesKeyAndSendClearDepartment(tcpClient, operationCodes.CLEAR_DEPARTMENT);
|
||||
if (success) {
|
||||
console.log(`Department directory cleared successfully for user ${userIp}`);
|
||||
} else {
|
||||
console.error(`Failed to clear department directory for user ${userIp}`);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`Error clearing department directory for user ${userIp}:`, error);
|
||||
} finally {
|
||||
tcpClient.closeSocket(); // Ensure the socket is closed
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for AES key to be set, send the file, and wait for the response
|
||||
private async waitForAesKeyAndSendFile(tcpClient: TcpClient, operationCode: string, metaInfo: { [key: string]: any }, fileContent: Buffer, checkInterval: number = 500, timeout: number = 10000): Promise<boolean> {
|
||||
const startTime = Date.now();
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const intervalId = setInterval(async () => {
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
// Check if AES key is set, if timeout occurs, reject
|
||||
if (!tcpClient.isAesKeySet()) {
|
||||
if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
console.error('Timeout waiting for AES key.');
|
||||
reject(new Error('Timeout waiting for AES key.'));
|
||||
}
|
||||
return; // Continue waiting for AES key
|
||||
}
|
||||
|
||||
// AES key is set, send the file
|
||||
clearInterval(intervalId);
|
||||
|
||||
try {
|
||||
const success = await tcpClient.sendMessage(operationCode, metaInfo, fileContent);
|
||||
if (!success) {
|
||||
reject(new Error('Failed to send the file content.'));
|
||||
return;
|
||||
}
|
||||
|
||||
// Wait for a response after sending the message
|
||||
const responseReceived = await this.waitForMessageResponse(tcpClient, timeout);
|
||||
if (!responseReceived) {
|
||||
reject(new Error('Timeout waiting for the message response.'));
|
||||
return;
|
||||
}
|
||||
|
||||
resolve(true); // Everything went fine
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
}, checkInterval); // Check for AES key every `checkInterval`
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for AES key to be set, send the clear department message, and wait for the response
|
||||
private async waitForAesKeyAndSendClearDepartment(tcpClient: TcpClient, operationCode: string, checkInterval: number = 500, timeout: number = 10000): Promise<boolean> {
|
||||
const startTime = Date.now();
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const intervalId = setInterval(async () => {
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
// Check if AES key is set, if timeout occurs, reject
|
||||
if (!tcpClient.isAesKeySet()) {
|
||||
if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
console.error('Timeout waiting for AES key.');
|
||||
reject(new Error('Timeout waiting for AES key.'));
|
||||
}
|
||||
return; // Continue waiting for AES key
|
||||
}
|
||||
|
||||
// AES key is set, send the clear department message
|
||||
clearInterval(intervalId);
|
||||
|
||||
try {
|
||||
const success = await tcpClient.sendMessage(operationCode, {});
|
||||
if (!success) {
|
||||
reject(new Error('Failed to send the clear department operation.'));
|
||||
return;
|
||||
}
|
||||
|
||||
// Wait for a response after sending the message
|
||||
const responseReceived = await this.waitForMessageResponse(tcpClient, timeout);
|
||||
if (!responseReceived) {
|
||||
reject(new Error('Timeout waiting for the message response.'));
|
||||
return;
|
||||
}
|
||||
|
||||
resolve(true); // Everything went fine
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
}, checkInterval); // Check for AES key every `checkInterval`
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for the response from the TCP client
|
||||
private waitForMessageResponse(tcpClient: TcpClient, timeout: number = 10000): Promise<boolean> {
|
||||
return new Promise((resolve) => {
|
||||
const startTime = Date.now();
|
||||
const intervalId = setInterval(() => {
|
||||
const response = tcpClient.getLastResult();
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
if (response?.operationCode === operationCodes.OK) {
|
||||
clearInterval(intervalId);
|
||||
resolve(true); // Message successfully received
|
||||
} else if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
resolve(false); // No response after timeout
|
||||
}
|
||||
}, 100); // Check every 100ms
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
import { promises as fs, watch, FSWatcher } from 'fs';
|
||||
import path from 'path';
|
||||
import { MemoryManager } from './memory_manager';
|
||||
import { JsonManager } from './json_manager';
|
||||
|
||||
export class DirectoryWatcher {
|
||||
private directoryPath: string;
|
||||
private directoryMemoryId: string;
|
||||
private directoryScheme: any;
|
||||
private applicationInfo: JsonManager;
|
||||
private memoryManager: MemoryManager;
|
||||
private readonly sourceKey: string;
|
||||
private directoryWatcher: FSWatcher | null; // To store the watcher reference
|
||||
private totalSize: number; // To store total directory size
|
||||
|
||||
constructor(memoryManagerPath: string, applicationInfoPath: string, sourceKey: string) {
|
||||
this.sourceKey = sourceKey;
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.memoryManager = new MemoryManager(memoryManagerPath);
|
||||
this.directoryMemoryId = '';
|
||||
this.directoryPath = '';
|
||||
this.directoryWatcher = null; // Initialize with no watcher
|
||||
this.totalSize = 0; // Initialize size with zero
|
||||
}
|
||||
|
||||
// Method to initialize and validate the backup directory
|
||||
async initialize(): Promise<boolean> {
|
||||
const directoryData = await this.applicationInfo.readValue(this.sourceKey);
|
||||
if (!directoryData) {
|
||||
return false;
|
||||
}
|
||||
|
||||
this.directoryPath = directoryData.path;
|
||||
this.directoryMemoryId = directoryData.id;
|
||||
|
||||
if (!this.directoryMemoryId || !this.directoryPath) {
|
||||
console.error('Components of entry in \'DirectoryWatcher\' not found.');
|
||||
await this.applicationInfo.removeValue(this.sourceKey);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!this.directoryScheme || Object.keys(this.directoryScheme).length === 0) {
|
||||
// No structure in memory, scan and save it
|
||||
const result = await this.buildDirectoryScheme(this.directoryPath);
|
||||
this.directoryScheme = result.structure;
|
||||
this.totalSize = result.size;
|
||||
|
||||
await this.memoryManager.updateMetaInformation(this.directoryMemoryId, {
|
||||
structure: this.directoryScheme,
|
||||
totalSize: this.totalSize,
|
||||
});
|
||||
}
|
||||
|
||||
// Start watching the directory (after stopping any existing watcher)
|
||||
this.restartWatcher();
|
||||
return true;
|
||||
}
|
||||
|
||||
// Recursively build the directory structure and calculate the total size
|
||||
private async buildDirectoryScheme(dirPath: string): Promise<{ structure: any, size: number }> {
|
||||
const directoryScheme: any = {};
|
||||
let totalSize = 0;
|
||||
|
||||
const items = await fs.readdir(dirPath, { withFileTypes: true });
|
||||
|
||||
for (const item of items) {
|
||||
const fullPath = path.join(dirPath, item.name);
|
||||
const stats = await fs.stat(fullPath); // Get stats for each item
|
||||
|
||||
if (item.isDirectory()) {
|
||||
// If it's a directory, recursively build its structure and accumulate size
|
||||
const { structure, size } = await this.buildDirectoryScheme(fullPath);
|
||||
directoryScheme[item.name] = structure;
|
||||
totalSize += size;
|
||||
} else if (item.isFile()) {
|
||||
// If it's a file, store its full path and accumulate size
|
||||
directoryScheme[item.name] = fullPath;
|
||||
totalSize += stats.size; // Add file size
|
||||
}
|
||||
}
|
||||
|
||||
return { structure: directoryScheme, size: totalSize };
|
||||
}
|
||||
|
||||
// Restart the directory watcher, ensuring any previous watcher is closed
|
||||
private restartWatcher(): void {
|
||||
if (this.directoryWatcher) {
|
||||
console.log('Stopping existing watcher...');
|
||||
this.directoryWatcher.close(); // Stop the existing watcher
|
||||
}
|
||||
|
||||
this.startDirectoryWatcher();
|
||||
}
|
||||
|
||||
// Start watching the backup directory for changes
|
||||
private startDirectoryWatcher(): void {
|
||||
if (!this.directoryPath) {
|
||||
throw new Error('Backup directory not set. Cannot start watcher.');
|
||||
}
|
||||
|
||||
this.directoryWatcher = watch(this.directoryPath, { recursive: true }, async (eventType, filename) => {
|
||||
if (filename) {
|
||||
console.log(`File change detected: ${eventType} - ${filename}`);
|
||||
// Rebuild the directory scheme and update memory
|
||||
const result = await this.buildDirectoryScheme(this.directoryPath);
|
||||
this.directoryScheme = result.structure;
|
||||
this.totalSize = result.size;
|
||||
|
||||
await this.memoryManager.updateMetaInformation(this.directoryMemoryId, {
|
||||
structure: this.directoryScheme,
|
||||
totalSize: this.totalSize,
|
||||
});
|
||||
|
||||
console.log('Directory structure and size updated in memory.');
|
||||
}
|
||||
});
|
||||
|
||||
console.log(`Watching for changes in: ${this.directoryPath}`);
|
||||
}
|
||||
|
||||
// Close the directory watcher
|
||||
public closeWatcher(): void {
|
||||
if (this.directoryWatcher) {
|
||||
console.log(`Stopping watcher for ${this.directoryPath}`);
|
||||
this.directoryWatcher.close();
|
||||
this.directoryWatcher = null; // Clear the reference after closing
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
import fs from 'fs';
|
||||
import crypto from 'crypto';
|
||||
|
||||
export class FileEncryptor {
|
||||
private readonly encryptionKey: Buffer;
|
||||
private readonly iv: Buffer;
|
||||
|
||||
constructor(base64Key: string, base64Iv: string) {
|
||||
// Decode the base64-encoded key and IV
|
||||
this.encryptionKey = Buffer.from(base64Key, 'base64');
|
||||
this.iv = Buffer.from(base64Iv, 'base64');
|
||||
}
|
||||
|
||||
// Method to read a file, encrypt it, and return the encrypted content as a base64 string
|
||||
public encryptFileToBase64(filePath: string): string {
|
||||
try {
|
||||
// Read the file contents
|
||||
const fileBuffer = fs.readFileSync(filePath);
|
||||
|
||||
// Create the cipher using AES-256-CBC (or another algorithm you prefer)
|
||||
const cipher = crypto.createCipheriv('aes-256-cbc', this.encryptionKey, this.iv);
|
||||
|
||||
// Encrypt the file data
|
||||
let encryptedData = cipher.update(fileBuffer);
|
||||
encryptedData = Buffer.concat([encryptedData, cipher.final()]);
|
||||
|
||||
// Return the encrypted data as a base64 string
|
||||
return encryptedData.toString('base64');
|
||||
} catch (err) {
|
||||
console.error(`Error encrypting file at path ${filePath}:`, err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// Method to decrypt base64-encoded encrypted content and return the decrypted buffer
|
||||
public decryptBase64(encryptedBase64: string): Buffer {
|
||||
try {
|
||||
// Decode the base64-encoded encrypted data
|
||||
const encryptedData = Buffer.from(encryptedBase64, 'base64');
|
||||
|
||||
// Create the decipher using AES-256-CBC
|
||||
const decipher = crypto.createDecipheriv('aes-256-cbc', this.encryptionKey, this.iv);
|
||||
|
||||
// Decrypt the data
|
||||
let decryptedData = decipher.update(encryptedData);
|
||||
decryptedData = Buffer.concat([decryptedData, decipher.final()]);
|
||||
|
||||
// Return the decrypted buffer
|
||||
return decryptedData;
|
||||
} catch (err) {
|
||||
console.error('Error decrypting data:', err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
import { QueueManager } from './queue_manager'; // Assume the QueueManager is in this path
|
||||
import { TcpClient } from '../network/tcp/tcp_client'; // Import your TcpClient class
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import fs from "fs";
|
||||
import path from "path";
|
||||
import {compareFnFileItemTask, FileItemTask} from "../interfaces/file_item_task";
|
||||
|
||||
interface FileSendTask {
|
||||
ip: string;
|
||||
path: string;
|
||||
userName: string;
|
||||
}
|
||||
|
||||
export class FileSharer {
|
||||
private queueManager: QueueManager<FileSendTask>;
|
||||
private readonly clientPort: number;
|
||||
|
||||
constructor(queueFilePath: string, clientPort: number) {
|
||||
|
||||
|
||||
this.queueManager = new QueueManager<FileItemTask>(queueFilePath, compareFnFileItemTask);
|
||||
this.clientPort = clientPort;
|
||||
}
|
||||
|
||||
// Start processing the file queue
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
await this.processQueue(); // Process the queue at regular intervals
|
||||
}, 10000); // 10 seconds interval
|
||||
}
|
||||
|
||||
// Method to process the queue
|
||||
private async processQueue(): Promise<void> {
|
||||
while (!this.queueManager.isEmpty()) {
|
||||
const task = this.queueManager.peek();
|
||||
|
||||
if (task) {
|
||||
const success = await this.sendFile(task);
|
||||
|
||||
if (!success) {
|
||||
console.error(`Failed to send file: ${task.path} to IP: ${task.ip}. Re-adding to queue.`);
|
||||
this.queueManager.dequeue();
|
||||
this.queueManager.enqueue(task); // Re-add to queue if failed
|
||||
}
|
||||
else{
|
||||
console.log(`File sent successfully: ${task.path} to IP: ${task.ip}.`);
|
||||
this.queueManager.dequeue();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Method to send the file to a specific IP using the TcpClient
|
||||
private async sendFile(task: FileSendTask): Promise<boolean> {
|
||||
const { ip, path: filePath, userName } = task;
|
||||
|
||||
// Ensure the file exists before attempting to send
|
||||
if (!fs.existsSync(filePath)) {
|
||||
console.error(`File not found: ${filePath}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Read the file contents
|
||||
const fileContent = fs.readFileSync(filePath);
|
||||
|
||||
// Extract the file name from the file path using path.basename (handles both Windows and Unix)
|
||||
const fileName = path.basename(filePath);
|
||||
|
||||
const metaInfo = {
|
||||
userName, // Sender's username
|
||||
relativeFilePath: fileName, // Use the file name instead of the full path
|
||||
};
|
||||
|
||||
const tcpClient = new TcpClient(this.clientPort);
|
||||
tcpClient.openSocket(ip);
|
||||
|
||||
// Wait for the AES key and then send the file
|
||||
try {
|
||||
const sendSuccess = await this.waitForAesKeyAndSendFile(tcpClient, operationCodes.SHARE_FILE, metaInfo, fileContent, task);
|
||||
tcpClient.closeSocket();
|
||||
|
||||
return sendSuccess;
|
||||
} catch (error) {
|
||||
console.error(`Error sending file: ${filePath} to IP: ${ip}. Error: ${error}`);
|
||||
tcpClient.closeSocket();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for AES key to be set and send the file, handling the response
|
||||
private async waitForAesKeyAndSendFile(tcpClient: TcpClient, operationCode: string, metaInfo: { [key: string]: any }, fileContent: Buffer, task: FileSendTask, checkInterval: number = 500, timeout: number = 10000): Promise<boolean> {
|
||||
const startTime = Date.now();
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const intervalId = setInterval(async () => {
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
// Check if AES key is set, if timeout occurs, reject
|
||||
if (!tcpClient.isAesKeySet()) {
|
||||
if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
console.error('Timeout waiting for AES key.');
|
||||
|
||||
reject(new Error('Timeout waiting for AES key.'));
|
||||
}
|
||||
return; // Continue waiting for AES key
|
||||
}
|
||||
|
||||
// AES key is set, send the message
|
||||
clearInterval(intervalId);
|
||||
|
||||
try {
|
||||
const success = await tcpClient.sendMessage(operationCode, metaInfo, fileContent);
|
||||
if (!success) {
|
||||
reject(new Error('Failed to send the file content.'));
|
||||
return;
|
||||
}
|
||||
|
||||
// Wait for a response after sending the message
|
||||
const responseReceived = await this.waitForMessageResponse(tcpClient, timeout);
|
||||
if (!responseReceived) {
|
||||
reject(new Error('Timeout waiting for the message response.'));
|
||||
return;
|
||||
}
|
||||
|
||||
resolve(true); // Everything went fine
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
}, checkInterval); // Check for AES key every `checkInterval`
|
||||
});
|
||||
}
|
||||
|
||||
// Method to wait for the response from the TCP client
|
||||
private waitForMessageResponse(tcpClient: TcpClient, timeout: number = 10000): Promise<boolean> {
|
||||
return new Promise((resolve) => {
|
||||
const startTime = Date.now();
|
||||
const intervalId = setInterval(() => {
|
||||
const response = tcpClient.getLastResult();
|
||||
const elapsedTime = Date.now() - startTime;
|
||||
|
||||
if (response?.operationCode === operationCodes.OK) {
|
||||
clearInterval(intervalId);
|
||||
resolve(true); // Message successfully received
|
||||
} else if (elapsedTime > timeout) {
|
||||
clearInterval(intervalId);
|
||||
resolve(false); // No response after timeout
|
||||
}
|
||||
}, 100); // Check every 100ms
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
import fs from 'fs';
|
||||
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 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
|
||||
if (!fs.existsSync(filePath)) {
|
||||
fs.writeFileSync(filePath, JSON.stringify({}, null, 2), 'utf8');
|
||||
}
|
||||
}
|
||||
|
||||
// Method to acquire a lock (create .lock file)
|
||||
private async acquireLock(): Promise<void> {
|
||||
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 (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<any | null> {
|
||||
await this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
if (!fs.existsSync(this.filePath)) return null;
|
||||
|
||||
const data = JSON.parse(fs.readFileSync(this.filePath, 'utf8'));
|
||||
return data[key] !== undefined ? data[key] : null;
|
||||
} catch (err: any) {
|
||||
console.error(`Error reading from JSON file: ${err.message}`);
|
||||
return null;
|
||||
} finally {
|
||||
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<boolean> {
|
||||
await this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
let data: { [key: string]: any } = {};
|
||||
|
||||
if (fs.existsSync(this.filePath)) {
|
||||
data = JSON.parse(fs.readFileSync(this.filePath, 'utf8'));
|
||||
}
|
||||
|
||||
// Update the key with the new value
|
||||
data[key] = value;
|
||||
|
||||
fs.writeFileSync(this.filePath, JSON.stringify(data, null, 2), 'utf8');
|
||||
return true;
|
||||
} catch (err: any) {
|
||||
console.error(`Error writing to JSON file: ${err.message}`);
|
||||
return false;
|
||||
} finally {
|
||||
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<boolean> {
|
||||
await this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
if (!fs.existsSync(this.filePath)) return false;
|
||||
|
||||
const data = JSON.parse(fs.readFileSync(this.filePath, 'utf8'));
|
||||
if (data[key] !== undefined) {
|
||||
delete data[key];
|
||||
fs.writeFileSync(this.filePath, JSON.stringify(data, null, 2), 'utf8');
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
} catch (err: any) {
|
||||
console.error(`Error removing key from JSON file: ${err.message}`);
|
||||
return false;
|
||||
} finally {
|
||||
this.releaseLock(); // Always release the lock after the operation
|
||||
}
|
||||
}
|
||||
|
||||
// Reset the JSON file by clearing all data with a lock
|
||||
public async resetFile(): Promise<boolean> {
|
||||
await this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
fs.writeFileSync(this.filePath, JSON.stringify({}, null, 2), 'utf8');
|
||||
return true;
|
||||
} catch (err: any) {
|
||||
console.error(`Error resetting JSON file: ${err.message}`);
|
||||
return false;
|
||||
} finally {
|
||||
this.releaseLock(); // Always release the lock after the operation
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
import { v4 as uuidv4 } from 'uuid';
|
||||
import { JsonManager } from './json_manager';
|
||||
|
||||
export class MemoryManager extends JsonManager {
|
||||
constructor(filePath: string) {
|
||||
super(filePath); // Call the parent constructor to ensure file initialization
|
||||
}
|
||||
|
||||
// Generate a new unique GUID and ensure it doesn't already exist in the file
|
||||
private generateUniqueGuid(): Promise<string> {
|
||||
const generate = async (): Promise<string> => {
|
||||
const guid = uuidv4();
|
||||
const value = await this.readValue(guid);
|
||||
if (value === null) {
|
||||
return guid;
|
||||
}
|
||||
return generate();
|
||||
};
|
||||
return generate();
|
||||
}
|
||||
|
||||
// Store meta information with a unique GUID as the key
|
||||
public async storeMetaInformation(metaInfo: any): Promise<string> {
|
||||
const guid = await this.generateUniqueGuid();
|
||||
const success = await this.writeValue(guid, metaInfo);
|
||||
if (success) {
|
||||
return guid; // Return the unique GUID for future reference
|
||||
} else {
|
||||
throw new Error('Failed to store meta information.');
|
||||
}
|
||||
}
|
||||
|
||||
// Retrieve meta information using the GUID
|
||||
public retrieveMetaInformation(guid: string): Promise<any | null> {
|
||||
return this.readValue(guid);
|
||||
}
|
||||
|
||||
// Update meta information by merging new data into existing data
|
||||
public async updateMetaInformation(guid: string, newMetaInfo: any): Promise<boolean> {
|
||||
return await this.writeValue(guid, newMetaInfo);
|
||||
}
|
||||
|
||||
// Remove meta information using the GUID
|
||||
public removeMetaInformation(guid: string): Promise<boolean> {
|
||||
return this.removeValue(guid);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
|
||||
export class QueueManager<T> {
|
||||
private readonly filePath: string;
|
||||
private readonly lockFilePath: string;
|
||||
private queue: T[];
|
||||
private readonly compareFn: (a: T, b: T) => boolean; // Comparison function
|
||||
|
||||
constructor(filePath: string, compareFn: (a: T, b: T) => boolean) {
|
||||
this.filePath = filePath;
|
||||
this.lockFilePath = `${filePath}.lock`; // Define the lock file path
|
||||
this.queue = [];
|
||||
this.compareFn = compareFn;
|
||||
|
||||
const dir = path.dirname(filePath);
|
||||
|
||||
// Check if the directory exists, throw error if it doesn't
|
||||
if (!fs.existsSync(dir)) {
|
||||
throw new Error(`The directory does not exist: ${dir}`);
|
||||
}
|
||||
|
||||
// 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 (create .lock file)
|
||||
private acquireLock(): void {
|
||||
while (fs.existsSync(this.lockFilePath)) {
|
||||
// Wait until the lock file is released
|
||||
this.sleepSync(100); // 100ms delay before retrying
|
||||
}
|
||||
// Create the lock file
|
||||
fs.writeFileSync(this.lockFilePath, '');
|
||||
}
|
||||
|
||||
// Sleep function to simulate delay for locking mechanism
|
||||
private sleepSync(ms: number): void {
|
||||
const start = Date.now();
|
||||
while (Date.now() - start < ms) {
|
||||
// busy wait
|
||||
}
|
||||
}
|
||||
|
||||
// Method to release the lock (delete .lock file)
|
||||
private releaseLock(): void {
|
||||
if (fs.existsSync(this.lockFilePath)) {
|
||||
fs.unlinkSync(this.lockFilePath);
|
||||
}
|
||||
}
|
||||
|
||||
// Load the queue from the JSON file
|
||||
loadQueue(): void {
|
||||
this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
const fileData = fs.readFileSync(this.filePath, 'utf8');
|
||||
this.queue = JSON.parse(fileData) || [];
|
||||
} catch (err) {
|
||||
// If the file doesn't exist or is invalid, start with an empty queue
|
||||
this.queue = [];
|
||||
} finally {
|
||||
this.releaseLock(); // Release the lock
|
||||
}
|
||||
}
|
||||
|
||||
// Save the queue back to the JSON file
|
||||
saveQueue(): void {
|
||||
this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
fs.writeFileSync(this.filePath, JSON.stringify(this.queue, null, 2), 'utf8');
|
||||
} finally {
|
||||
this.releaseLock(); // Release the lock
|
||||
}
|
||||
}
|
||||
|
||||
// Enqueue: Add an item to the end of the queue if it doesn't already exist
|
||||
enqueue(item: T): void {
|
||||
this.loadQueue(); // Ensure we load the latest queue
|
||||
|
||||
// Check if the item already exists in the queue
|
||||
const exists = this.queue.some(existingItem => this.compareFn(existingItem, item));
|
||||
|
||||
console.log(this.queue);
|
||||
|
||||
if (!exists) {
|
||||
this.queue.push(item);
|
||||
this.saveQueue(); // Save the updated queue
|
||||
} else {
|
||||
console.log('Item already exists in the queue. Skipping enqueue.');
|
||||
}
|
||||
}
|
||||
|
||||
// Dequeue: Remove an item from the front of the queue
|
||||
dequeue(): T | null {
|
||||
this.loadQueue(); // Ensure we load the latest queue
|
||||
if (this.queue.length === 0) {
|
||||
return null; // Queue is empty
|
||||
}
|
||||
const item = this.queue.shift() as T; // Remove the first item
|
||||
this.saveQueue(); // Save the updated queue
|
||||
return item;
|
||||
}
|
||||
|
||||
// Peek: Get the item at the front of the queue without removing it
|
||||
peek(): T | null {
|
||||
this.loadQueue(); // Ensure we load the latest queue
|
||||
return this.queue.length > 0 ? this.queue[0] : null;
|
||||
}
|
||||
|
||||
// Check if the queue is empty
|
||||
isEmpty(): boolean {
|
||||
this.loadQueue(); // Ensure we load the latest queue
|
||||
return this.queue.length === 0;
|
||||
}
|
||||
|
||||
// Get the length of the queue
|
||||
length(): number {
|
||||
this.loadQueue(); // Ensure we load the latest queue
|
||||
return this.queue.length;
|
||||
}
|
||||
|
||||
// Clear the entire queue
|
||||
clearQueue(): void {
|
||||
this.acquireLock(); // Acquire the lock
|
||||
|
||||
try {
|
||||
this.queue = []; // Clear the queue
|
||||
fs.writeFileSync(this.filePath, JSON.stringify(this.queue, null, 2), 'utf8');
|
||||
} finally {
|
||||
this.releaseLock(); // Release the lock
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
import { JsonManager } from './json_manager'; // Assuming you have this class for managing user/memory
|
||||
import { WindowManager } from './window_manager'; // For managing app navigation
|
||||
import { UdpClient } from '../network/udp/udp_client';
|
||||
|
||||
export class TaskScheduler {
|
||||
private applicationInfo: JsonManager;
|
||||
private windowManager: WindowManager;
|
||||
private appStarted: boolean = false;
|
||||
private intervalIds: NodeJS.Timeout[] = []; // Array to store interval IDs
|
||||
|
||||
constructor(applicationInfo: JsonManager, windowManager: WindowManager) {
|
||||
this.applicationInfo = applicationInfo;
|
||||
this.windowManager = windowManager;
|
||||
}
|
||||
|
||||
// Function to schedule the UC check task with dynamic UDP client creation
|
||||
public startUCCheck(udpPort: number, okPage: string, errorPage: string, interval: number = 30000) {
|
||||
const intervalId = setInterval(async () => {
|
||||
try {
|
||||
const udpClient = new UdpClient(udpPort); // Create UdpClient with the provided port
|
||||
const aliveClients = await udpClient.getAliveClients();
|
||||
const storedIp = await this.applicationInfo.readValue('serverIp');
|
||||
const foundClient = aliveClients.length > 0;
|
||||
|
||||
if (foundClient) {
|
||||
const ipAddress = aliveClients[0]; // Just using the first alive client
|
||||
|
||||
if (!storedIp || storedIp !== ipAddress) {
|
||||
await this.applicationInfo.writeValue('serverIp', ipAddress);
|
||||
if (!this.appStarted) {
|
||||
await this.windowManager.changeContent(okPage);
|
||||
}
|
||||
this.appStarted = true;
|
||||
} else if (!this.appStarted) {
|
||||
await this.windowManager.changeContent(okPage);
|
||||
this.appStarted = true;
|
||||
}
|
||||
} else {
|
||||
await this.windowManager.changeContent(errorPage);
|
||||
}
|
||||
} catch (err) {
|
||||
console.error('Error checking UC:', err);
|
||||
await this.windowManager.changeContent(errorPage);
|
||||
}
|
||||
}, interval);
|
||||
|
||||
this.intervalIds.push(intervalId);
|
||||
}
|
||||
|
||||
// Function to schedule the IP lookup task, storing the active addresses in memory
|
||||
public async startUserIPLookup(udpPort: number, interval: number = 30000) {
|
||||
const serverIp = await this.applicationInfo.readValue('serverIp'); // Ensure it's awaited if it's an async function
|
||||
const intervalId = setInterval(async () => {
|
||||
try {
|
||||
const udpClient = new UdpClient(udpPort); // Create a UDP client with the provided port
|
||||
const activeIPs = await udpClient.getAliveClients(); // Get the list of active IPs
|
||||
|
||||
// Filter out the serverIp from the list of active clients
|
||||
const filteredIPs = activeIPs.filter(ip => ip !== serverIp);
|
||||
|
||||
// Save the filtered IPs to 'users_ip'
|
||||
await this.applicationInfo.writeValue('users_ip', filteredIPs);
|
||||
} catch (err) {
|
||||
console.error('Error during user IP lookup:', err);
|
||||
}
|
||||
}, interval);
|
||||
|
||||
this.intervalIds.push(intervalId); // Store the interval ID
|
||||
}
|
||||
|
||||
// Function to stop all tasks (UC check, user worker, etc.)
|
||||
public stopAllTasks() {
|
||||
this.intervalIds.forEach(clearInterval); // Clear all intervals
|
||||
this.intervalIds = []; // Reset the array
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
import { TcpClient } from "../network/tcp/tcp_client";
|
||||
import { ParsedMessage } from "../network/message_handler";
|
||||
|
||||
export class TcpCommunicator {
|
||||
private readonly ip: string;
|
||||
private readonly port: number;
|
||||
private tcpClient: TcpClient | null = null;
|
||||
private lastResult: ParsedMessage | null = null;
|
||||
|
||||
constructor(ip: string, port: number) {
|
||||
this.ip = ip;
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
async connect(): Promise<boolean> {
|
||||
this.tcpClient = new TcpClient(this.port);
|
||||
this.tcpClient.openSocket(this.ip);
|
||||
return this.tcpClient.isSocketConnected();
|
||||
}
|
||||
|
||||
async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<boolean> {
|
||||
if (!this.tcpClient || !this.tcpClient.isSocketConnected()) return false;
|
||||
|
||||
// Wait until the AES key is set before sending the message
|
||||
return new Promise((resolve) => {
|
||||
const idWaitForAes = setInterval(async () => {
|
||||
if (this.tcpClient?.isAesKeySet()) {
|
||||
clearInterval(idWaitForAes);
|
||||
|
||||
// Send the message once AES key is set
|
||||
const status = await this.tcpClient!.sendMessage(operationCode, metaInfo, fileContent);
|
||||
|
||||
if (status) {
|
||||
await this.waitForResponse();
|
||||
}
|
||||
|
||||
resolve(status);
|
||||
}
|
||||
}, 100);
|
||||
});
|
||||
}
|
||||
|
||||
getLastResult(): ParsedMessage | null {
|
||||
const message = this.lastResult;
|
||||
this.lastResult = null;
|
||||
return message;
|
||||
}
|
||||
|
||||
hasResponseArrived(): boolean {
|
||||
if(!this.tcpClient) return false;
|
||||
return this.lastResult !== null;
|
||||
}
|
||||
|
||||
private waitForResponse(): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (!this.tcpClient) {
|
||||
reject();
|
||||
}
|
||||
|
||||
// Start interval for waiting for the response
|
||||
const responseInterval = setInterval(() => {
|
||||
if (!this.tcpClient?.isSocketConnected()) {
|
||||
clearInterval(responseInterval);
|
||||
resolve();
|
||||
}
|
||||
|
||||
if (this.tcpClient?.isMessageReceived()) {
|
||||
this.lastResult = this.tcpClient.getLastResult();
|
||||
clearInterval(responseInterval); // Stop checking once we have a response
|
||||
resolve();
|
||||
}
|
||||
}, 100); // Check every 100 milliseconds
|
||||
});
|
||||
}
|
||||
|
||||
async disconnect(): Promise<boolean> {
|
||||
if (!this.tcpClient || !this.tcpClient.isSocketConnected()) return true;
|
||||
this.tcpClient.closeSocket();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
import { JsonManager } from "./json_manager";
|
||||
import { MemoryManager } from "./memory_manager";
|
||||
import { operationCodes } from "../network/operation_codes";
|
||||
import { TcpCommunicator } from "./tcp_communicator";
|
||||
|
||||
export class UsersInfoFetcher {
|
||||
private applicationInfo: JsonManager;
|
||||
private memoryManager: MemoryManager;
|
||||
private tcpCommunicator: TcpCommunicator | null = null;
|
||||
private readonly clientPort: number;
|
||||
private memoryId: string;
|
||||
private readonly activeUsersKey: string;
|
||||
|
||||
constructor(applicationInfoPath: string, memoryManagerPath: string, clientPort: number) {
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.memoryManager = new MemoryManager(memoryManagerPath);
|
||||
this.memoryId = '';
|
||||
this.clientPort = clientPort;
|
||||
this.tcpCommunicator = null;
|
||||
this.activeUsersKey = 'active_users_info';
|
||||
}
|
||||
|
||||
// Method to start checking user info periodically (every minute)
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
await this.initialize(); // Re-run every minute
|
||||
}, 60000); // 1 minute interval
|
||||
}
|
||||
|
||||
// Initialize and fetch user IPs and process users info
|
||||
private async initialize() {
|
||||
const usersIps = await this.applicationInfo.readValue('users_ip');
|
||||
if (!usersIps) {
|
||||
console.error('No IP addresses found in users_ip');
|
||||
return;
|
||||
}
|
||||
|
||||
// Ensure active_users_info exists in the memory
|
||||
this.memoryId = await this.applicationInfo.readValue(this.activeUsersKey);
|
||||
if (!this.memoryId) {
|
||||
this.memoryId = await this.memoryManager.storeMetaInformation([]);
|
||||
await this.applicationInfo.writeValue(this.activeUsersKey, this.memoryId);
|
||||
}
|
||||
|
||||
// Check user information
|
||||
await this.checkUsersInfo(usersIps);
|
||||
}
|
||||
|
||||
// Check user info from the list of IPs
|
||||
private async checkUsersInfo(usersIps: string[]) {
|
||||
let usersInfo: Array<{ ip: string, user_info: any }> = []; // Array to store IP and user_info objects
|
||||
|
||||
for (const ip of usersIps) {
|
||||
this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort);
|
||||
if (!await this.tcpCommunicator.connect()) {
|
||||
console.error(`Failed to open connection for IP: ${ip}`);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Wait for the response for 10 seconds
|
||||
const response = await this.waitForResponse();
|
||||
console.log(response);
|
||||
|
||||
// If a response is received and is successful, append it to usersInfo
|
||||
if (response && response.metaInfo) {
|
||||
console.log(`Response received from ${ip}:`, response);
|
||||
usersInfo.push({
|
||||
ip: ip,
|
||||
user_info: response.metaInfo // Push the IP and user_info object into the array
|
||||
});
|
||||
}
|
||||
|
||||
await this.tcpCommunicator.disconnect();
|
||||
}
|
||||
|
||||
await this.updateActiveUsers(usersInfo); // Update active users information in the memory
|
||||
}
|
||||
|
||||
// Send the message once and wait for the response for a specific timeout
|
||||
private async waitForResponse(timeout: number = 10000): Promise<any> {
|
||||
// Send the message once
|
||||
const status = await this.tcpCommunicator?.sendMessage(operationCodes.GET_USER_INFORMATION);
|
||||
if (!status) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Wait for the response until the timeout
|
||||
return new Promise((resolve) => {
|
||||
const startTime = Date.now();
|
||||
|
||||
const checkResponseInterval = setInterval(async () => {
|
||||
// Check if the message has been received
|
||||
if (this.tcpCommunicator?.hasResponseArrived()) {
|
||||
clearInterval(checkResponseInterval);
|
||||
resolve(this.tcpCommunicator?.getLastResult()); // Return the response once it arrives
|
||||
}
|
||||
|
||||
// If the timeout is reached, stop checking and resolve with null
|
||||
if (Date.now() - startTime > timeout) {
|
||||
clearInterval(checkResponseInterval);
|
||||
resolve(null);
|
||||
}
|
||||
}, 100); // Check every 100ms if the response has arrived
|
||||
});
|
||||
}
|
||||
|
||||
// Update active users information in the memory
|
||||
private async updateActiveUsers(userInfo: any[]) {
|
||||
await this.memoryManager.updateMetaInformation(this.memoryId, userInfo); // Update active users in memory
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
import { BrowserWindow, dialog, shell } from 'electron';
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
|
||||
export class WindowManager {
|
||||
private readonly mainWindow: BrowserWindow;
|
||||
private readonly pathToPagesDir: string;
|
||||
|
||||
constructor(mainWindow: BrowserWindow, pathToPagesDir: string) {
|
||||
this.pathToPagesDir = pathToPagesDir;
|
||||
this.mainWindow = mainWindow;
|
||||
}
|
||||
|
||||
// Show an alert dialog
|
||||
async showAlert(message: string): Promise<void> {
|
||||
if (this.mainWindow) {
|
||||
await dialog.showMessageBox(this.mainWindow, {
|
||||
type: 'info',
|
||||
title: 'Alert',
|
||||
message: message,
|
||||
buttons: ['OK'],
|
||||
});
|
||||
} else {
|
||||
console.error('Main window is not available.');
|
||||
}
|
||||
}
|
||||
|
||||
// Change the content of the current window to load a new HTML file
|
||||
async changeContent(destination: string): Promise<void> {
|
||||
if (this.mainWindow) {
|
||||
try {
|
||||
const destinationPath = path.join(this.pathToPagesDir, `${destination}.html`);
|
||||
console.log(`Navigating to: ${destinationPath}`);
|
||||
|
||||
// Load the destination HTML file into the main window
|
||||
await this.mainWindow.loadFile(destinationPath);
|
||||
console.log(`Navigated to ${destination}`);
|
||||
} catch (error) {
|
||||
console.error('Error changing content:', error);
|
||||
throw error; // Pass the error back to the render process
|
||||
}
|
||||
} else {
|
||||
console.error('Main window is not available.');
|
||||
}
|
||||
}
|
||||
|
||||
// New method to select a directory
|
||||
async selectDirectory(): Promise<string | undefined> {
|
||||
const result = await dialog.showOpenDialog(this.mainWindow, {
|
||||
properties: ['openDirectory'], // Only allow selecting directories
|
||||
});
|
||||
|
||||
// If the user cancels, result.filePaths will be an empty array
|
||||
if (result.filePaths && result.filePaths.length > 0) {
|
||||
return result.filePaths[0]; // Return the selected directory path
|
||||
} else {
|
||||
console.log('No directory selected.');
|
||||
return undefined; // Return undefined if no directory was selected
|
||||
}
|
||||
}
|
||||
|
||||
async showFileInExplorer(filePath: string): Promise<void> {
|
||||
if (filePath && fs.existsSync(filePath)) {
|
||||
try {
|
||||
// Use Electron's shell module to show the file in the explorer
|
||||
shell.showItemInFolder(filePath);
|
||||
console.log(`Opened file explorer for: ${filePath}`);
|
||||
} catch (error: any) {
|
||||
console.error(`Error showing file in explorer: ${error.message}`);
|
||||
}
|
||||
} else {
|
||||
console.error('File path is undefined or does not exist.');
|
||||
}
|
||||
}
|
||||
|
||||
// New method to open the file explorer and choose a file
|
||||
async selectFile(): Promise<string | undefined> {
|
||||
const result = await dialog.showOpenDialog(this.mainWindow, {
|
||||
properties: ['openFile'], // Allow selecting a file
|
||||
filters: [
|
||||
{ name: 'All Files', extensions: ['*'] } // Optionally filter for specific file types
|
||||
]
|
||||
});
|
||||
|
||||
if (result.filePaths && result.filePaths.length > 0) {
|
||||
return result.filePaths[0]; // Return the selected file path
|
||||
} else {
|
||||
console.log('No file selected.');
|
||||
return undefined; // Return undefined if no file was selected
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,155 @@
|
||||
import { Worker } from 'worker_threads';
|
||||
import path from 'path';
|
||||
import {WindowManager} from "./window_manager";
|
||||
|
||||
export class WorkerManager {
|
||||
private readonly pathToWorkerDir: string;
|
||||
private windowManager: WindowManager
|
||||
private workers: Worker[]; // Array to store running workers
|
||||
|
||||
constructor(pathToWorkerDir: string, windowManager: WindowManager) {
|
||||
this.pathToWorkerDir = pathToWorkerDir;
|
||||
this.windowManager = windowManager;
|
||||
this.workers = []; // Initialize the array to store workers
|
||||
}
|
||||
|
||||
// Start the Connection Pool Worker
|
||||
async startWatchersWorker(memoryManagerPath: string, applicationInfoPath: string): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const worker = new Worker(path.join(this.pathToWorkerDir, '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('Connection Pool Worker message:', data);
|
||||
});
|
||||
|
||||
worker.on('error', (err) => {
|
||||
console.error('Connection Pool Worker error:', err);
|
||||
worker.terminate();
|
||||
this.removeWorker(worker);
|
||||
reject(err); // Reject the promise if there's an error
|
||||
});
|
||||
|
||||
worker.on('exit', (code) => {
|
||||
console.log(`Connection Pool Worker exited with code ${code}`);
|
||||
this.removeWorker(worker); // Remove worker reference when it exits
|
||||
resolve(); // Resolve when the worker exits cleanly
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Start the Servers Worker (UDP and TCP servers)
|
||||
async startServersWorker(host: string, udpPort: number, tcpPort: number): Promise<void> {
|
||||
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
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Start the Users Info Worker
|
||||
async startResourceCoordinatorWorker(usersConfigPath: string, applicationInfoPath: string, memoryManagerPath: string, queueManagerPath: string, tcpPort: number): Promise<void> {
|
||||
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
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Start the Backup Retrieval Worker
|
||||
async startBackupRetrievalWorker(
|
||||
userConfigPath: string,
|
||||
applicationInfoPath: string,
|
||||
clientPort: number,
|
||||
destinationPath: string
|
||||
): Promise<void> {
|
||||
return new Promise((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', (data) => {
|
||||
console.log('Backup Retrieval Worker message:', data);
|
||||
this.windowManager.changeContent('main_menu');
|
||||
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
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Close all running workers
|
||||
closeAllWorkers(): void {
|
||||
console.log('Terminating all running workers...');
|
||||
this.workers.forEach(worker => worker.terminate()); // Terminate each worker
|
||||
this.workers = []; // Clear the array after terminating all workers
|
||||
}
|
||||
|
||||
// Helper method to remove a worker from the workers array when it exits
|
||||
private removeWorker(worker: Worker): void {
|
||||
const index = this.workers.indexOf(worker);
|
||||
if (index > -1) {
|
||||
this.workers.splice(index, 1); // Remove the worker from the array
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user