From fa4750ea3bbba316e5e62c268ba7c826d495e03c Mon Sep 17 00:00:00 2001 From: andrei-mihnea-cerbu Date: Tue, 12 Nov 2024 23:14:08 +0200 Subject: [PATCH] network chunk v2 --- User/src/helpers/file_sharer.ts | 12 ++--- .../user_to_user_operations.ts | 2 + .../tcp_client_communicator.ts | 44 ++++++++++--------- .../tcp_server_communicator.ts | 44 ++++++++++--------- 4 files changed, 53 insertions(+), 49 deletions(-) diff --git a/User/src/helpers/file_sharer.ts b/User/src/helpers/file_sharer.ts index 68e254d..7b688c4 100644 --- a/User/src/helpers/file_sharer.ts +++ b/User/src/helpers/file_sharer.ts @@ -28,24 +28,18 @@ export class FileSharer { async start(): Promise { setInterval(async () => { if (!this.isBusy) { // Check if the queue is already being processed + this.isBusy = true; // Set busy flag to true before starting this.log("Start successfully. Processing the queue."); - await this.processQueue(); // Process the queue at regular intervals - this.log("Queue processing completed."); + await this.processQueue(); } }, 10000); // 10 seconds interval } // Method to process the queue private async processQueue(): Promise { - if (this.isBusy) { - this.log("Queue is already being processed. Skipping this interval."); - return; - } - - this.isBusy = true; // Set busy flag to true before starting - while (!this.queueManager.isEmpty()) { const task = this.queueManager.peek(); + this.log('trimiti fisier'); if (task) { this.log(`Processing task for file: ${task.path} to IP: ${task.ip}`); diff --git a/User/src/network/operations_custom/user_to_user_operations.ts b/User/src/network/operations_custom/user_to_user_operations.ts index adefe62..9cc8dec 100644 --- a/User/src/network/operations_custom/user_to_user_operations.ts +++ b/User/src/network/operations_custom/user_to_user_operations.ts @@ -116,6 +116,8 @@ export class UserToUserOperations implements OperationPlugin { const appInfo = await jsonManager.readValue('shareDirectory'); const shareDirectory = appInfo?.path || ''; + console.log(`\n\nShare directory: ${shareDirectory}\n\n`); + if (!shareDirectory) { return { operationCode: operationCodes.ERR, metaInfo: { message: 'Share directory missing.' }}; } diff --git a/User/src/network/socket_communicator/tcp_client_communicator.ts b/User/src/network/socket_communicator/tcp_client_communicator.ts index 5bffe31..1305457 100644 --- a/User/src/network/socket_communicator/tcp_client_communicator.ts +++ b/User/src/network/socket_communicator/tcp_client_communicator.ts @@ -113,37 +113,41 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { } async handleIncomingChunk(data: Buffer): Promise { - const incomingMessage = data.toString(); - this.messageBuffer += incomingMessage; + const incomingMessage = data.toString().trim(); - if (this.messageBuffer.includes(END_OF_MESSAGE)) { - const messages = this.messageBuffer.split(END_OF_MESSAGE); - - for (let i = 0; i < messages.length - 1; i++) { - const completeMessage = messages[i]; - if (completeMessage) { - await this.processCompleteMessage(completeMessage); - } - } - - this.messageBuffer = messages[messages.length - 1]; + // Check if the message is the end marker + if (incomingMessage === END_OF_MESSAGE) { + // Process the complete message when EOM is received + await this.processCompleteMessage(); + return; } - } - private async processCompleteMessage(completeMessage: string): Promise { - const [headerJson, chunkContent] = completeMessage.split('|'); + // Extract header and chunk content from incoming data + const [headerJson, chunkContent] = incomingMessage.split('|'); const header = JSON.parse(headerJson); + // Initialize chunk array if this is the first chunk for this messageId if (!this.chunkBuffers[header.messageId]) { this.chunkBuffers[header.messageId] = new Array(header.totalChunks); } + // Place the chunk in the correct position in the chunk buffer this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent; + } - if (this.chunkBuffers[header.messageId].every((chunk) => chunk !== undefined)) { - const fullMessage = this.chunkBuffers[header.messageId].join(''); - await this.handleIncomingMessage(fullMessage); - delete this.chunkBuffers[header.messageId]; + // Process the complete message when EOM is received + private async processCompleteMessage(): Promise { + for (const [messageId, chunks] of Object.entries(this.chunkBuffers)) { + if (chunks.every((chunk) => chunk !== undefined)) { + // Join all chunks to form the full message + const fullMessage = chunks.join(''); + + // Handle the completed and possibly decrypted message + await this.handleIncomingMessage(fullMessage); + + // Clean up buffer after processing + delete this.chunkBuffers[messageId]; + } } } diff --git a/User/src/network/socket_communicator/tcp_server_communicator.ts b/User/src/network/socket_communicator/tcp_server_communicator.ts index feb1632..4f03d2c 100644 --- a/User/src/network/socket_communicator/tcp_server_communicator.ts +++ b/User/src/network/socket_communicator/tcp_server_communicator.ts @@ -97,37 +97,41 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { // Handle incoming chunks of data async handleIncomingChunk(data: Buffer): Promise { - const incomingMessage = data.toString(); - this.messageBuffer += incomingMessage; + const incomingMessage = data.toString().trim(); - if (this.messageBuffer.includes(END_OF_MESSAGE)) { - const messages = this.messageBuffer.split(END_OF_MESSAGE); - - for (let i = 0; i < messages.length - 1; i++) { - const completeMessage = messages[i]; - if (completeMessage) { - await this.processCompleteMessage(completeMessage); - } - } - - this.messageBuffer = messages[messages.length - 1]; + // Check if the message is the end marker + if (incomingMessage === END_OF_MESSAGE) { + // Process the complete message when EOM is received + await this.processCompleteMessage(); + return; } - } - private async processCompleteMessage(completeMessage: string): Promise { - const [headerJson, chunkContent] = completeMessage.split('|'); + // Extract header and chunk content from incoming data + const [headerJson, chunkContent] = incomingMessage.split('|'); const header = JSON.parse(headerJson); + // Initialize chunk array if this is the first chunk for this messageId if (!this.chunkBuffers[header.messageId]) { this.chunkBuffers[header.messageId] = new Array(header.totalChunks); } + // Place the chunk in the correct position in the chunk buffer this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent; + } - if (this.chunkBuffers[header.messageId].every((chunk) => chunk !== undefined)) { - const fullMessage = this.chunkBuffers[header.messageId].join(''); - await this.handleIncomingMessage(fullMessage); - delete this.chunkBuffers[header.messageId]; + // Process the complete message when EOM is received + private async processCompleteMessage(): Promise { + for (const [messageId, chunks] of Object.entries(this.chunkBuffers)) { + if (chunks.every((chunk) => chunk !== undefined)) { + // Join all chunks to form the full message + const fullMessage = chunks.join(''); + + // Handle the completed and possibly decrypted message + await this.handleIncomingMessage(fullMessage); + + // Clean up buffer after processing + delete this.chunkBuffers[messageId]; + } } }