From c83481b6d4910d6ff2c877bb3d78c252bb7c709b Mon Sep 17 00:00:00 2001 From: andrei-mihnea-cerbu Date: Tue, 12 Nov 2024 14:03:37 +0200 Subject: [PATCH] refactor v2 --- CEO/src/helpers/announcement_sender.ts | 84 +++++++++++++++++++++ CEO/src/helpers/worker_manager.ts | 29 +++++++ CEO/src/workers/send_announcement_worker.ts | 19 +++++ 3 files changed, 132 insertions(+) create mode 100644 CEO/src/helpers/announcement_sender.ts create mode 100644 CEO/src/workers/send_announcement_worker.ts diff --git a/CEO/src/helpers/announcement_sender.ts b/CEO/src/helpers/announcement_sender.ts new file mode 100644 index 0000000..c1f1fd5 --- /dev/null +++ b/CEO/src/helpers/announcement_sender.ts @@ -0,0 +1,84 @@ +import { JsonManager } from './json_manager'; // Assuming this manages JSON configurations +import { TcpCommunicator } from "./tcp_communicator"; +import { operationCodes } from '../network/operation_codes'; // Assuming this holds operation codes +import { parentPort } from 'worker_threads'; +import {ParsedMessage} from "../network/message_handler"; + +export class AnnouncementSender { + private applicationInfo: JsonManager; + private readonly clientPort: number; + private message: string = ''; + private tcpCommunicator: TcpCommunicator | null = null; + + constructor(applicationInfoPath: string, clientPort: number) { + this.applicationInfo = new JsonManager(applicationInfoPath); + this.clientPort = clientPort; + } + + async start(message: string): Promise { + console.log('AnnouncementWorker started.'); + this.message = message; + try { + const activeUsersIp = await this.applicationInfo.readValue('users_ip'); + if (!activeUsersIp || !activeUsersIp.length) { + throw new Error('No active users found.'); + } + + for (const ip of activeUsersIp) { + const success = await this.sendAnnouncementToIp(ip); + if (!success) { + throw new Error(`Failed to send announcement to all users.`); + } + console.log(`Announcement sent and confirmed successfully from ${ip}`); + } + + parentPort?.postMessage({ success: true, message: 'Announcement sent to all active users successfully.' }); + } catch (error: any) { + console.error('Error in AnnouncementWorker:', error); + parentPort?.postMessage({ success: false, message: `A problem occurred: ${error.message}` }); + } + + console.log('AnnouncementWorker finished.'); + } + + private async sendAnnouncementToIp(ip: string): Promise { + this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort); + if (!await this.tcpCommunicator.connect()) { + console.log(`Skipping user at IP ${ip} - unable to connect.`); + return true; + } + + // Prepare the message metadata + const metaInfo = { message: this.message }; + + // Send the announcement message + const messageSent = await this.tcpCommunicator.sendMessage(operationCodes.SEND_ANNOUNCEMENT, metaInfo); + if (!messageSent) { + await this.tcpCommunicator.disconnect(); + return false; + } + + // Await confirmation from the user + const response = await this.waitForResponse(); + if (response?.operationCode === operationCodes.OK) { + await this.tcpCommunicator.disconnect(); + return true; + } + + // If confirmation is not OK, disconnect and halt + await this.tcpCommunicator.disconnect(); + return false; + } + + private async waitForResponse(): Promise { + return new Promise((resolve) => { + const idResponseCheck = setInterval(async () => { + if (!this.tcpCommunicator) return null; + if (this.tcpCommunicator.hasResponseArrived()) { + clearInterval(idResponseCheck); + resolve(this.tcpCommunicator.getLastResult()); + } + }, 100); + }); + } +} diff --git a/CEO/src/helpers/worker_manager.ts b/CEO/src/helpers/worker_manager.ts index f53f43d..a2401bc 100644 --- a/CEO/src/helpers/worker_manager.ts +++ b/CEO/src/helpers/worker_manager.ts @@ -168,6 +168,35 @@ export class WorkerManager { }); } + async startAnnouncementWorker(applicationInfoPath: string, clientPort: number, message: string): Promise { + return new Promise((resolve, reject) => { + const worker = new Worker(path.join(this.pathToWorkerDir, 'send_announcement_worker.js'), { + workerData: { applicationInfoPath, clientPort, message }, // Pass necessary data to the worker + }); + + this.workers.push(worker); // Store the worker reference + + worker.on('message', async (data) => { + await new Promise((resolve) => setTimeout(resolve, 3000)); // Wait for 1 second + this.windowManager.changeContent('main_menu'); + this.windowManager.showAlert(`${data.message}`) + }); + + worker.on('error', (err) => { + console.error('Announcement Worker error:', err); + worker.terminate(); + this.removeWorker(worker); + reject(err); // Reject the promise if there's an error + }); + + worker.on('exit', (code) => { + console.log(`Announcement Worker exited with code ${code}`); + this.removeWorker(worker); // Remove worker reference when it exits + resolve(); // Resolve when the worker exits cleanly + }); + }); + } + // Close all running workers closeAllWorkers(): void { console.log('Terminating all running workers...'); diff --git a/CEO/src/workers/send_announcement_worker.ts b/CEO/src/workers/send_announcement_worker.ts new file mode 100644 index 0000000..816b6e7 --- /dev/null +++ b/CEO/src/workers/send_announcement_worker.ts @@ -0,0 +1,19 @@ +import { workerData } from 'worker_threads'; +import { AnnouncementSender} from "../helpers/announcement_sender"; + +// Destructure data passed from the main thread +const { + applicationInfoPath, + clientPort, + message +}: { + applicationInfoPath: string, + clientPort: number, + message: string +} = workerData; + +// Initialize the AnnouncementWorker +const announcementWorker = new AnnouncementSender(applicationInfoPath, clientPort); + +// Start the announcement process and handle results +announcementWorker.start(message);