overall v1
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
import { JsonManager } from './json_manager'; // Assuming this manages JSON configurations
|
||||
import { JsonManager } from './json_manager';
|
||||
import { TcpCommunicator } from "./tcp_communicator";
|
||||
import { operationCodes } from '../network/operation_codes'; // Assuming this holds operation codes
|
||||
import { parentPort } from 'worker_threads';
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import path from 'path';
|
||||
import fs from 'fs';
|
||||
import crypto from 'crypto';
|
||||
@@ -15,15 +14,30 @@ export class BackupRetrievalWorker {
|
||||
private encryptionKey: Buffer | null = null;
|
||||
private iv: Buffer | null = null;
|
||||
private tcpCommunicator: TcpCommunicator | null = null;
|
||||
private isBusy: boolean;
|
||||
private lastProcessedUserIndex: number;
|
||||
|
||||
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;
|
||||
this.isBusy = false;
|
||||
this.lastProcessedUserIndex = 0;
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
if (!this.isBusy) {
|
||||
this.log('Start successfully. Processing backup tasks.');
|
||||
await this.processBackupTasks();
|
||||
}
|
||||
}, 10000); // Retry every 10 seconds if there's an error
|
||||
}
|
||||
|
||||
private async processBackupTasks(): Promise<void> {
|
||||
this.isBusy = true;
|
||||
|
||||
try {
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.name) {
|
||||
@@ -41,46 +55,50 @@ export class BackupRetrievalWorker {
|
||||
|
||||
const activeUsersIp = await this.applicationInfo.readValue('users_ip');
|
||||
if (!activeUsersIp || !activeUsersIp.length) {
|
||||
parentPort?.postMessage({ success: false, message: 'No active users found.' });
|
||||
return;
|
||||
this.isBusy = false;
|
||||
throw new Error('No active users found.');
|
||||
}
|
||||
|
||||
// Process each user, starting from the last processed index
|
||||
let backupSuccessful = true;
|
||||
for (const ip of activeUsersIp) {
|
||||
for (let i = this.lastProcessedUserIndex; i < activeUsersIp.length; i++) {
|
||||
const ip = activeUsersIp[i];
|
||||
const success = await this.processBackupForIp(ip, userName);
|
||||
if (!success) {
|
||||
backupSuccessful = false;
|
||||
parentPort?.postMessage({ success: false, message: 'Backup could not be completed due to an internal error.' });
|
||||
this.lastProcessedUserIndex = i; // Remember where it stopped
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (backupSuccessful) {
|
||||
parentPort?.postMessage({ success: true, message: 'Backup successful.' });
|
||||
this.log('Backup retrieval completed successfully for all users.');
|
||||
this.lastProcessedUserIndex = 0; // Reset index to start from the beginning next cycle
|
||||
}
|
||||
} catch (error: any) {
|
||||
console.error('Error in BackupRetrievalWorker:', error);
|
||||
parentPort?.postMessage({ success: false, message: `Backup could not be completed due to an internal error: ${error.message}` });
|
||||
this.log(error.message, 'error');
|
||||
}
|
||||
|
||||
this.isBusy = false;
|
||||
}
|
||||
|
||||
private async processBackupForIp(ip: string, userName: string): Promise<boolean> {
|
||||
this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort);
|
||||
if (!await this.tcpCommunicator.connect()) {
|
||||
console.error(`Failed to connect to ${ip}`);
|
||||
this.log(`Failed to connect to ${ip}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
const backupExists = await this.checkIfBackupExists(userName);
|
||||
if (!backupExists) {
|
||||
console.log(`No backup found for user ${userName} on IP ${ip}`);
|
||||
this.log(`No backup found for user ${userName} on IP ${ip}`);
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true; // Skip user if no backup found, do not mark as error
|
||||
}
|
||||
|
||||
const backupStructure = await this.requestBackupStructure(userName);
|
||||
if (!backupStructure || Object.keys(backupStructure).length === 0) {
|
||||
console.log(`No files found in backup structure for user ${userName} on IP ${ip}`);
|
||||
this.log(`No files found in backup structure for user ${userName} on IP ${ip}`);
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true; // Skip user if no files found, do not mark as error
|
||||
}
|
||||
@@ -88,7 +106,7 @@ export class BackupRetrievalWorker {
|
||||
for (const relativeFilePath of Object.keys(backupStructure)) {
|
||||
const fileRequestSuccess = await this.requestBackupFile(userName, relativeFilePath);
|
||||
if (!fileRequestSuccess) {
|
||||
console.error(`Failed to retrieve file ${relativeFilePath} from backup for user ${userName} on IP ${ip}`);
|
||||
this.log(`Failed to retrieve file ${relativeFilePath} from backup for user ${userName} on IP ${ip}`, 'error');
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return false; // Stop if any file fails to be retrieved
|
||||
}
|
||||
@@ -133,7 +151,7 @@ export class BackupRetrievalWorker {
|
||||
|
||||
private saveFile(relativeFilePath: string, fileContent: string): boolean {
|
||||
if (!this.encryptionKey || !this.iv) {
|
||||
console.error('Encryption key or IV is not set.');
|
||||
this.log('Encryption key or IV is not set.', 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -141,7 +159,7 @@ export class BackupRetrievalWorker {
|
||||
try {
|
||||
encryptedBuffer = Buffer.from(fileContent, 'base64');
|
||||
} catch (error) {
|
||||
console.error('Error decoding base64 file content:', error);
|
||||
this.log(`Error decoding base64 file content: ${error}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -150,7 +168,7 @@ export class BackupRetrievalWorker {
|
||||
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);
|
||||
this.log(`Error decrypting file: ${error}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -161,10 +179,10 @@ export class BackupRetrievalWorker {
|
||||
fs.mkdirSync(dirPath, { recursive: true });
|
||||
}
|
||||
fs.writeFileSync(fullFilePath, decryptedContent);
|
||||
console.log(`File saved successfully: ${fullFilePath}`);
|
||||
this.log(`File saved successfully: ${fullFilePath}`);
|
||||
return true;
|
||||
} catch (error: any) {
|
||||
console.error(`Error saving file ${relativeFilePath}: ${error.message}`);
|
||||
this.log(`Error saving file ${relativeFilePath}: ${error.message}`, 'error');
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -177,7 +195,17 @@ export class BackupRetrievalWorker {
|
||||
clearInterval(idResponseCheck);
|
||||
resolve(this.tcpCommunicator.getLastResult());
|
||||
}
|
||||
}, 100);
|
||||
}, 100); // Check every 100 milliseconds
|
||||
});
|
||||
}
|
||||
|
||||
// Unified logging function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const prefix = '[BackupRetrievalWorker]';
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
import { JsonManager } from './json_manager'; // Assuming this manages JSON configurations
|
||||
import { JsonManager } from './json_manager';
|
||||
import { TcpCommunicator } from "./tcp_communicator";
|
||||
import { operationCodes } from '../network/operation_codes'; // Assuming this holds operation codes
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import { parentPort } from 'worker_threads';
|
||||
import path from 'path';
|
||||
import fs from 'fs';
|
||||
import crypto from 'crypto';
|
||||
import { ParsedMessage } from "../network/message_handler"; // Import the crypto module for encryption and decryption
|
||||
import { ParsedMessage } from "../network/message_handler";
|
||||
|
||||
export class BackupRetrievalWorker {
|
||||
private userConfig: JsonManager;
|
||||
@@ -23,6 +23,16 @@ export class BackupRetrievalWorker {
|
||||
this.destinationPath = destinationPath;
|
||||
}
|
||||
|
||||
// Logging helper function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const prefix = '[BackupRetrievalWorker]';
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
try {
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
@@ -49,12 +59,12 @@ export class BackupRetrievalWorker {
|
||||
if (!success) {
|
||||
throw new Error(`Failed to retrieve backup from ${ip}`);
|
||||
}
|
||||
console.log(`Backup retrieved successfully from ${ip}`);
|
||||
this.log(`Backup retrieved successfully from ${ip}`);
|
||||
}
|
||||
|
||||
parentPort?.postMessage({ success: true, message: 'Backup retrieval completed successfully.' });
|
||||
} catch (error: any) {
|
||||
console.error('Error in BackupRetrievalWorker:', error);
|
||||
this.log(`Error in BackupRetrievalWorker: ${error.message}`, 'error');
|
||||
parentPort?.postMessage({ success: false, message: `A problem occurred: ${error.message}` });
|
||||
}
|
||||
}
|
||||
@@ -62,19 +72,20 @@ export class BackupRetrievalWorker {
|
||||
private async processBackupForIp(ip: string, userName: string): Promise<boolean> {
|
||||
this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort);
|
||||
if (!await this.tcpCommunicator.connect()) {
|
||||
this.log(`Failed to connect to ${ip}`, 'error');
|
||||
return true;
|
||||
}
|
||||
|
||||
const backupExists = await this.checkIfBackupExists(userName);
|
||||
if (!backupExists) {
|
||||
console.log(`No backup found for user ${userName} on IP ${ip}`);
|
||||
this.log(`No backup found for user ${userName} on IP ${ip}`);
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true;
|
||||
}
|
||||
|
||||
const backupStructure = await this.requestBackupStructure(userName);
|
||||
if (!backupStructure || Object.keys(backupStructure).length === 0) {
|
||||
console.log(`No files found in backup structure for user ${userName} on IP ${ip}`);
|
||||
this.log(`No files found in backup structure for user ${userName} on IP ${ip}`);
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true;
|
||||
}
|
||||
@@ -151,7 +162,7 @@ export class BackupRetrievalWorker {
|
||||
fs.mkdirSync(dirPath, { recursive: true });
|
||||
}
|
||||
fs.writeFileSync(fullFilePath, decryptedContent);
|
||||
console.log(`File saved successfully: ${fullFilePath}`);
|
||||
this.log(`File saved successfully: ${fullFilePath}`);
|
||||
return true;
|
||||
} catch (error: any) {
|
||||
throw new Error(`Error saving file ${relativeFilePath}: ${error.message}`);
|
||||
|
||||
@@ -1,19 +1,19 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { TcpCommunicator } from './tcp_communicator'; // Using TcpCommunicator instead of TcpClient
|
||||
import { TcpCommunicator } from './tcp_communicator';
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import { JsonManager } from './json_manager'; // Manages JSON configurations
|
||||
import { MemoryManager } from './memory_manager'; // Manages in-memory data structures
|
||||
import { JsonManager } from './json_manager';
|
||||
import { MemoryManager } from './memory_manager';
|
||||
import { ParsedMessage } from "../network/message_handler";
|
||||
|
||||
export class DepartmentSharer {
|
||||
private userConfig: JsonManager;
|
||||
private applicationInfo: JsonManager;
|
||||
private memoryManager: MemoryManager; // To read the department files
|
||||
private memoryManager: MemoryManager;
|
||||
private departmentDirectory: string | null;
|
||||
private readonly clientPort: number;
|
||||
private isBusy: boolean = false; // Busy flag to prevent simultaneous sharing
|
||||
private tcpCommunicator: TcpCommunicator | null = null; // For each user connection
|
||||
private isBusy: boolean = false;
|
||||
private tcpCommunicator: TcpCommunicator | null = null;
|
||||
|
||||
constructor(
|
||||
userConfigPath: string,
|
||||
@@ -32,21 +32,20 @@ export class DepartmentSharer {
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
if (!this.isBusy) {
|
||||
this.isBusy = true;
|
||||
this.log('Start successfully. Sharing files with the department.');
|
||||
await this.shareFilesWithDepartment();
|
||||
this.isBusy = false;
|
||||
}
|
||||
}, 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');
|
||||
this.isBusy = true;
|
||||
|
||||
// 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.');
|
||||
this.log('User information or department ID is missing in the configuration.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -56,27 +55,27 @@ export class DepartmentSharer {
|
||||
// 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.');
|
||||
this.log('No active users found.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
const activeUsers = await this.memoryManager.retrieveMetaInformation(activeUsersId);
|
||||
if (!activeUsers || activeUsers.length === 0) {
|
||||
console.error('No active users found.');
|
||||
this.log('No active users found.', 'error');
|
||||
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.');
|
||||
this.log('No users found in the same department.', 'log');
|
||||
return;
|
||||
}
|
||||
|
||||
// Get department directory info
|
||||
const departmentData = await this.applicationInfo.readValue('departmentDirectory');
|
||||
if (!departmentData || !departmentData.path || !departmentData.id) {
|
||||
console.error('No department directory found.');
|
||||
this.log('No department directory found.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -85,7 +84,7 @@ export class DepartmentSharer {
|
||||
// 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.');
|
||||
this.log('No files found for this department in the memory manager.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -114,7 +113,7 @@ export class DepartmentSharer {
|
||||
const response = await this.waitForResponse();
|
||||
|
||||
if (!response || response.operationCode !== operationCodes.OK){
|
||||
console.error('Failed to clear the department directory.');
|
||||
this.log('Failed to clear the department directory.', 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -124,14 +123,14 @@ export class DepartmentSharer {
|
||||
// Send the files to a user in the department
|
||||
private async sendFilesToUser(files: { [key: string]: string }, userName: string): Promise<void> {
|
||||
if(!this.tcpCommunicator) return;
|
||||
const unsentFiles = Object.keys(files); // Keep track of unsent files
|
||||
const unsentFiles = Object.keys(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}`);
|
||||
this.log(`File not found: ${filePath}`, 'error');
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -144,8 +143,8 @@ export class DepartmentSharer {
|
||||
|
||||
// Prepare the metaInfo (same structure as FileSharer)
|
||||
const metaInfo = {
|
||||
userName, // Sender's username
|
||||
relativeFilePath // Use the relative path to preserve directory structure
|
||||
userName,
|
||||
relativeFilePath
|
||||
};
|
||||
|
||||
// Send the file
|
||||
@@ -153,7 +152,7 @@ export class DepartmentSharer {
|
||||
|
||||
const response = await this.waitForResponse();
|
||||
if (!response || response.operationCode !== operationCodes.OK) {
|
||||
console.error(`Failed to send file: ${fileName}`);
|
||||
this.log(`Failed to send file: ${fileName}`, 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -168,9 +167,19 @@ export class DepartmentSharer {
|
||||
if (!this.tcpCommunicator) return null;
|
||||
if (this.tcpCommunicator.hasResponseArrived()) {
|
||||
clearInterval(idResponseCheck);
|
||||
resolve(this.tcpCommunicator.getLastResult()); // Resolve the response or null if not available
|
||||
resolve(this.tcpCommunicator.getLastResult());
|
||||
}
|
||||
}, 100); // Check every 100 milliseconds if the response has arrived
|
||||
}, 100);
|
||||
});
|
||||
}
|
||||
|
||||
// Unified logging function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const prefix = '[DepartmentSharer]';
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10,8 +10,9 @@ export class DirectoryWatcher {
|
||||
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
|
||||
private directoryWatcher: FSWatcher | null;
|
||||
private totalSize: number;
|
||||
private isBusy: boolean;
|
||||
|
||||
constructor(memoryManagerPath: string, applicationInfoPath: string, sourceKey: string) {
|
||||
this.sourceKey = sourceKey;
|
||||
@@ -19,14 +20,28 @@ export class DirectoryWatcher {
|
||||
this.memoryManager = new MemoryManager(memoryManagerPath);
|
||||
this.directoryMemoryId = '';
|
||||
this.directoryPath = '';
|
||||
this.directoryWatcher = null; // Initialize with no watcher
|
||||
this.totalSize = 0; // Initialize size with zero
|
||||
this.directoryWatcher = null;
|
||||
this.totalSize = 0;
|
||||
this.isBusy = false;
|
||||
}
|
||||
|
||||
// Start the watcher with a busy flag to prevent overlapping operations
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
if (!this.isBusy) {
|
||||
this.isBusy = true;
|
||||
const initialized = await this.initialize();
|
||||
if (initialized) this.log('Directory watcher started successfully.');
|
||||
this.isBusy = false;
|
||||
}
|
||||
}, 10000); // 10-second interval for testing
|
||||
}
|
||||
|
||||
// Method to initialize and validate the backup directory
|
||||
async initialize(): Promise<boolean> {
|
||||
const directoryData = await this.applicationInfo.readValue(this.sourceKey);
|
||||
if (!directoryData) {
|
||||
this.log('Directory data not found in application info.', 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -34,7 +49,7 @@ export class DirectoryWatcher {
|
||||
this.directoryMemoryId = directoryData.id;
|
||||
|
||||
if (!this.directoryMemoryId || !this.directoryPath) {
|
||||
console.error('Components of entry in \'DirectoryWatcher\' not found.');
|
||||
this.log('Components of entry in \'DirectoryWatcher\' not found.', 'error');
|
||||
await this.applicationInfo.removeValue(this.sourceKey);
|
||||
return false;
|
||||
}
|
||||
@@ -65,7 +80,7 @@ export class DirectoryWatcher {
|
||||
|
||||
for (const item of items) {
|
||||
const fullPath = path.join(dirPath, item.name);
|
||||
const stats = await fs.stat(fullPath); // Get stats for each item
|
||||
const stats = await fs.stat(fullPath);
|
||||
|
||||
if (item.isDirectory()) {
|
||||
// If it's a directory, recursively build its structure and accumulate size
|
||||
@@ -75,7 +90,7 @@ export class DirectoryWatcher {
|
||||
} 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
|
||||
totalSize += stats.size;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -85,8 +100,8 @@ export class DirectoryWatcher {
|
||||
// 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.log('Stopping existing watcher...');
|
||||
this.directoryWatcher.close();
|
||||
}
|
||||
|
||||
this.startDirectoryWatcher();
|
||||
@@ -100,7 +115,7 @@ export class DirectoryWatcher {
|
||||
|
||||
this.directoryWatcher = watch(this.directoryPath, { recursive: true }, async (eventType, filename) => {
|
||||
if (filename) {
|
||||
console.log(`File change detected: ${eventType} - ${filename}`);
|
||||
this.log(`File change detected: ${eventType} - ${filename}`);
|
||||
// Rebuild the directory scheme and update memory
|
||||
const result = await this.buildDirectoryScheme(this.directoryPath);
|
||||
this.directoryScheme = result.structure;
|
||||
@@ -111,19 +126,34 @@ export class DirectoryWatcher {
|
||||
totalSize: this.totalSize,
|
||||
});
|
||||
|
||||
console.log('Directory structure and size updated in memory.');
|
||||
this.log('Directory structure and size updated in memory.');
|
||||
}
|
||||
});
|
||||
|
||||
console.log(`Watching for changes in: ${this.directoryPath}`);
|
||||
this.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.log(`Stopping watcher for ${this.directoryPath}`);
|
||||
this.directoryWatcher.close();
|
||||
this.directoryWatcher = null; // Clear the reference after closing
|
||||
this.directoryWatcher = null;
|
||||
}
|
||||
}
|
||||
|
||||
// Unified logging function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const sourcePrefix = `[DirectoryWatcher] {${this.capitalize(this.sourceKey)}}`;
|
||||
if (level === 'error') {
|
||||
console.error(`${sourcePrefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${sourcePrefix} ${message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Capitalize the first letter of the sourceKey
|
||||
private capitalize(str: string): string {
|
||||
return str.charAt(0).toUpperCase() + str.slice(1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import {QueueManager} from './queue_manager';
|
||||
import {TcpCommunicator} from "./tcp_communicator"; // Updated to use TcpCommunicator
|
||||
import {operationCodes} from '../network/operation_codes';
|
||||
import fs from "fs";
|
||||
import path from "path";
|
||||
import {compareFnFileItemTask, FileItemTask} from "../interfaces/file_item_task";
|
||||
import {ParsedMessage} from "../network/message_handler";
|
||||
import { QueueManager } from './queue_manager';
|
||||
import { TcpCommunicator } from "./tcp_communicator";
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import { compareFnFileItemTask, FileItemTask } from "../interfaces/file_item_task";
|
||||
import { ParsedMessage } from "../network/message_handler";
|
||||
|
||||
interface FileSendTask {
|
||||
ip: string;
|
||||
@@ -28,7 +28,9 @@ export class FileSharer {
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
if (!this.isBusy) { // Check if the queue is already being processed
|
||||
this.log("Start successfully. Processing the queue.");
|
||||
await this.processQueue(); // Process the queue at regular intervals
|
||||
this.log("Queue processing completed.");
|
||||
}
|
||||
}, 10000); // 10 seconds interval
|
||||
}
|
||||
@@ -36,7 +38,7 @@ export class FileSharer {
|
||||
// Method to process the queue
|
||||
private async processQueue(): Promise<void> {
|
||||
if (this.isBusy) {
|
||||
console.log("Queue is already being processed. Skipping this interval.");
|
||||
this.log("Queue is already being processed. Skipping this interval.");
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -46,15 +48,15 @@ export class FileSharer {
|
||||
const task = this.queueManager.peek();
|
||||
|
||||
if (task) {
|
||||
console.log(task);
|
||||
this.log(`Processing task for file: ${task.path} to IP: ${task.ip}`);
|
||||
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.log(`Failed to send file: ${task.path} to IP: ${task.ip}. Re-adding to queue.`, 'error');
|
||||
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.log(`File sent successfully: ${task.path} to IP: ${task.ip}`);
|
||||
this.queueManager.dequeue();
|
||||
}
|
||||
}
|
||||
@@ -64,11 +66,11 @@ export class FileSharer {
|
||||
|
||||
// Method to send the file to a specific IP using TcpCommunicator
|
||||
private async sendFile(task: FileSendTask): Promise<boolean> {
|
||||
const {ip, path: filePath, userName} = task;
|
||||
const { ip, path: filePath, userName } = task;
|
||||
|
||||
// Ensure the file exists before attempting to send
|
||||
if (!fs.existsSync(filePath)) {
|
||||
console.error(`File not found: ${filePath}`);
|
||||
this.log(`File not found: ${filePath}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -86,17 +88,17 @@ export class FileSharer {
|
||||
this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort);
|
||||
|
||||
if (!await this.tcpCommunicator.connect()) {
|
||||
console.error(`Failed to connect to IP: ${ip}`);
|
||||
this.log(`Failed to connect to IP: ${ip}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
console.log(`Sending file: ${filePath} to IP: ${ip}`);
|
||||
this.log(`Sending file: ${filePath} to IP: ${ip}`);
|
||||
|
||||
if (!await this.tcpCommunicator.sendMessage(operationCodes.SHARE_FILE, metaInfo, fileContent)) return false;
|
||||
|
||||
const response = await this.waitForResponse();
|
||||
if (!response || response.operationCode !== operationCodes.OK) {
|
||||
console.error(`Failed to send file: ${filePath} to IP: ${ip}`);
|
||||
this.log(`Failed to send file: ${filePath} to IP: ${ip}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -115,4 +117,14 @@ export class FileSharer {
|
||||
}, 100); // Check every 100 milliseconds if the response has arrived
|
||||
});
|
||||
}
|
||||
|
||||
// Unified logging function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const prefix = '[FileSharer]';
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
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";
|
||||
|
||||
export class NetworkScanner {
|
||||
private applicationInfo: JsonManager;
|
||||
private userConfig: JsonManager;
|
||||
private readonly udpPort: number;
|
||||
private readonly tcpPort: number;
|
||||
private readonly okPage: string;
|
||||
private readonly errorPage: string;
|
||||
private readonly databaseResetPage: string;
|
||||
private appStarted = false;
|
||||
private intervalIds: NodeJS.Timeout[] = [];
|
||||
|
||||
// Flags to prevent overlapping executions
|
||||
private ucCheckBusy = false;
|
||||
private ipLookupBusy = false;
|
||||
private sendLoginBusy = false;
|
||||
|
||||
constructor(applicationInfoPath: string, userConfigPath: string, udpPort: number, tcpPort: number, okPage: string, errorPage: string, databaseResetPage: string) {
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.userConfig = new JsonManager(userConfigPath);
|
||||
this.udpPort = udpPort;
|
||||
this.tcpPort = tcpPort;
|
||||
this.okPage = okPage;
|
||||
this.errorPage = errorPage;
|
||||
this.databaseResetPage = databaseResetPage;
|
||||
|
||||
// Start tasks
|
||||
this.startUCCheck();
|
||||
this.startUserIPLookup();
|
||||
this.sendLoginRequest();
|
||||
}
|
||||
|
||||
// Log helper function for consistent logging format
|
||||
private log(message: string, level: 'log' | 'error' = 'log', methodName: string = ''): void {
|
||||
const prefix = `[NetworkScanner${methodName ? `.${methodName}` : ''}]`;
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
}
|
||||
|
||||
// UC Check Task
|
||||
private startUCCheck(interval: number = 5000): void {
|
||||
const intervalId = setInterval(async () => {
|
||||
if (this.ucCheckBusy) return;
|
||||
this.ucCheckBusy = true;
|
||||
|
||||
try {
|
||||
this.log('UC Check running...', 'log', 'startUCCheck');
|
||||
const udpClient = new UdpClient(this.udpPort);
|
||||
const aliveClients = await udpClient.getAliveClients();
|
||||
const storedIp = await this.applicationInfo.readValue('serverIp');
|
||||
const foundClient = aliveClients.length > 0;
|
||||
|
||||
if (foundClient) {
|
||||
const ipAddress = aliveClients[0]; // Use the first alive client
|
||||
|
||||
if (!storedIp || storedIp !== ipAddress) {
|
||||
await this.applicationInfo.writeValue('serverIp', ipAddress);
|
||||
if (!this.appStarted) {
|
||||
parentPort?.postMessage({ type: 'changeContent', page: this.okPage });
|
||||
}
|
||||
this.appStarted = true;
|
||||
} else if (!this.appStarted) {
|
||||
parentPort?.postMessage({ type: 'changeContent', page: this.okPage });
|
||||
this.appStarted = true;
|
||||
}
|
||||
} else {
|
||||
parentPort?.postMessage({ type: 'changeContent', page: this.errorPage });
|
||||
}
|
||||
} catch (err) {
|
||||
this.log(`Error checking UC: ${err}`, 'error', 'startUCCheck');
|
||||
parentPort?.postMessage({ type: 'changeContent', page: this.errorPage });
|
||||
} finally {
|
||||
this.ucCheckBusy = false;
|
||||
}
|
||||
}, interval);
|
||||
|
||||
this.intervalIds.push(intervalId);
|
||||
}
|
||||
|
||||
// IP Lookup Task
|
||||
private startUserIPLookup(interval: number = 10000): void {
|
||||
const intervalId = setInterval(async () => {
|
||||
if (this.ipLookupBusy) return;
|
||||
this.ipLookupBusy = true;
|
||||
|
||||
try {
|
||||
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 filteredIPs = activeIPs.filter(ip => ip !== serverIp);
|
||||
|
||||
// Save the filtered IPs to 'users_ip'
|
||||
await this.applicationInfo.writeValue('users_ip', filteredIPs);
|
||||
} catch (err) {
|
||||
this.log(`Error during user IP lookup: ${err}`, 'error', 'startUserIPLookup');
|
||||
} finally {
|
||||
this.ipLookupBusy = false;
|
||||
}
|
||||
}, interval);
|
||||
|
||||
this.intervalIds.push(intervalId);
|
||||
}
|
||||
|
||||
// Login Request Task
|
||||
private sendLoginRequest(interval: number = 5000): void {
|
||||
const intervalId = setInterval(async () => {
|
||||
if (this.sendLoginBusy || this.appStarted) return;
|
||||
this.sendLoginBusy = true;
|
||||
|
||||
try {
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.email || !userInfo.password) {
|
||||
this.log("Email or password not found in user config.", 'error', 'sendLoginRequest');
|
||||
return;
|
||||
}
|
||||
|
||||
const app_type = await this.userConfig.readValue('app_type');
|
||||
const email = userInfo.email;
|
||||
const password = userInfo.password;
|
||||
const serverIp = await this.applicationInfo.readValue('serverIp');
|
||||
|
||||
if (!serverIp) {
|
||||
this.log("Server IP not found in application info.", 'error', 'sendLoginRequest');
|
||||
return;
|
||||
}
|
||||
|
||||
const tcpCommunicator = new TcpCommunicator(serverIp, this.tcpPort);
|
||||
|
||||
if (!await tcpCommunicator.connect()) {
|
||||
this.log("Failed to connect to the server.", 'error', 'sendLoginRequest');
|
||||
return;
|
||||
}
|
||||
|
||||
const metaInfo = { email, password, app_type };
|
||||
if (!await tcpCommunicator.sendMessage(operationCodes.LOGIN, metaInfo)) {
|
||||
this.log("Failed to send login request.", 'error', 'sendLoginRequest');
|
||||
await tcpCommunicator.disconnect();
|
||||
return;
|
||||
}
|
||||
|
||||
const response = await this.waitForResponse(tcpCommunicator);
|
||||
if (response?.operationCode !== operationCodes.OK) {
|
||||
await this.userConfig.resetFile();
|
||||
await this.userConfig.writeValue('app_type', app_type);
|
||||
parentPort?.postMessage({ type: 'changeContent', page: this.databaseResetPage });
|
||||
}
|
||||
} catch (err) {
|
||||
this.log(`Error during login request: ${err}`, 'error', 'sendLoginRequest');
|
||||
} finally {
|
||||
this.sendLoginBusy = false;
|
||||
}
|
||||
}, interval);
|
||||
|
||||
this.intervalIds.push(intervalId);
|
||||
}
|
||||
|
||||
// Helper function to wait for a response from the TCP communicator
|
||||
private async waitForResponse(tcpCommunicator: TcpCommunicator): Promise<ParsedMessage | null> {
|
||||
return new Promise((resolve) => {
|
||||
const checkInterval = setInterval(() => {
|
||||
if (tcpCommunicator.hasResponseArrived()) {
|
||||
clearInterval(checkInterval);
|
||||
resolve(tcpCommunicator.getLastResult());
|
||||
}
|
||||
}, 100);
|
||||
});
|
||||
}
|
||||
|
||||
// Method to stop all intervals (for cleanup if needed)
|
||||
public stopAllIntervals(): void {
|
||||
for (const id of this.intervalIds) {
|
||||
clearInterval(id);
|
||||
}
|
||||
this.log("All intervals have been stopped.", 'log', 'stopAllIntervals');
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,7 @@ import { JsonManager } from "./json_manager";
|
||||
import { MemoryManager } from "./memory_manager";
|
||||
import { operationCodes } from "../network/operation_codes";
|
||||
import { TcpCommunicator } from "./tcp_communicator";
|
||||
import {ParsedMessage} from "../network/message_handler";
|
||||
import { ParsedMessage } from "../network/message_handler";
|
||||
|
||||
export class UsersInfoFetcher {
|
||||
private applicationInfo: JsonManager;
|
||||
@@ -32,7 +32,7 @@ export class UsersInfoFetcher {
|
||||
private async initialize() {
|
||||
const usersIps = await this.applicationInfo.readValue('users_ip');
|
||||
if (!usersIps) {
|
||||
console.error('No IP addresses found in users_ip');
|
||||
this.log('No IP addresses found in users_ip', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -54,13 +54,13 @@ export class UsersInfoFetcher {
|
||||
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}`);
|
||||
this.log(`Failed to open connection for IP: ${ip}`, 'error');
|
||||
continue;
|
||||
}
|
||||
|
||||
if(!await this.tcpCommunicator?.sendMessage(operationCodes.GET_USER_INFORMATION)){
|
||||
await this.tcpCommunicator.disconnect();
|
||||
continue
|
||||
continue;
|
||||
}
|
||||
|
||||
// Wait for the response for 10 seconds
|
||||
@@ -96,4 +96,14 @@ export class UsersInfoFetcher {
|
||||
private async updateActiveUsers(userInfo: any[]) {
|
||||
await this.memoryManager.updateMetaInformation(this.memoryId, userInfo); // Update active users in memory
|
||||
}
|
||||
|
||||
// Unified logging function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const prefix = '[UsersInfoFetcher]';
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,7 +22,6 @@ export class WorkerManager {
|
||||
this.workers.push(worker); // Store the worker reference
|
||||
|
||||
worker.on('message', (data) => {
|
||||
console.log(data);
|
||||
if (data.type === 'changeContent') {
|
||||
this.windowManager.changeContent(data.page);
|
||||
}
|
||||
@@ -43,7 +42,7 @@ export class WorkerManager {
|
||||
});
|
||||
}
|
||||
|
||||
// Start the Connection Pool Worker
|
||||
// Start the Directories Watcher Worker
|
||||
async startDirectoriesWatchersWorker(memoryManagerPath: string, applicationInfoPath: string): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const worker = new Worker(path.join(this.pathToWorkerDir, 'directories_watcher_worker.js'), {
|
||||
@@ -53,18 +52,18 @@ export class WorkerManager {
|
||||
this.workers.push(worker); // Store the worker reference
|
||||
|
||||
worker.on('message', (data) => {
|
||||
console.log('Connection Pool Worker message:', data);
|
||||
console.log('DirectoriesWatcher message:', data);
|
||||
});
|
||||
|
||||
worker.on('error', (err) => {
|
||||
console.error('Connection Pool Worker 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(`Connection Pool Worker exited with code ${code}`);
|
||||
console.log(`DirectoriesWatcher exited with code ${code}`);
|
||||
this.removeWorker(worker); // Remove worker reference when it exits
|
||||
resolve(); // Resolve when the worker exits cleanly
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user