From 60e71c3e2aa8dd75cce2245fba912e40c2ef92e3 Mon Sep 17 00:00:00 2001 From: andrei-mihnea-cerbu Date: Wed, 13 Nov 2024 09:45:14 +0200 Subject: [PATCH] network chunk v6 --- .../tcp_server_communicator.ts | 29 +++++++++++++++---- .../tcp_client_communicator.ts | 21 +++++++++++--- .../tcp_server_communicator.ts | 26 +++++++++++++---- 3 files changed, 62 insertions(+), 14 deletions(-) diff --git a/UC/src/network/socket_communicator/tcp_server_communicator.ts b/UC/src/network/socket_communicator/tcp_server_communicator.ts index 89741ba..fe60b59 100644 --- a/UC/src/network/socket_communicator/tcp_server_communicator.ts +++ b/UC/src/network/socket_communicator/tcp_server_communicator.ts @@ -4,7 +4,7 @@ import { MessageHandler } from '../message_handler'; import { SocketCommunicatorBase } from './socket_communicator_base'; import { OperationHandler } from '../operations_base/operation_handler'; -const CHUNK_SIZE = 1024; // Define chunk size +const MAX_CHUNK_SIZE = 2048; // Define chunk size export class TcpServerCommunicator extends SocketCommunicatorBase { private readonly socket: Socket; @@ -100,6 +100,8 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { const [headerJson, chunkContent] = incomingMessage.split('|'); const header = JSON.parse(headerJson); + console.log(`\n\n${incomingMessage}\n\n`); + // 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); @@ -129,13 +131,18 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { } // Send chunked message - async sendChunkedMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise { + async sendChunkedMessage( + operationCode: string, + metaInfo?: { [key: string]: any }, + fileContent?: Buffer + ): Promise { const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); let outgoingMessage: string; + // Encrypt or format message based on the operation code switch(operationCode) { case 'SET_PUBLIC_KEY': - outgoingMessage = message; + outgoingMessage = Buffer.from(message, 'utf-8').toString('base64'); break; case 'SET_AES_KEY': outgoingMessage = this.encryptWithRsa(message); @@ -144,11 +151,21 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { outgoingMessage = this.encryptWithAes(message); } - const totalChunks = Math.ceil(outgoingMessage.length / CHUNK_SIZE); + // Determine optimal chunk size (multiple of 4 and ≤ 1024 to align with base64 encoding) + let optimalChunkSize = MAX_CHUNK_SIZE; + while (outgoingMessage.length % optimalChunkSize !== 0 && optimalChunkSize > 0) { + optimalChunkSize -= 4; + } + + if (optimalChunkSize === 0) optimalChunkSize = MAX_CHUNK_SIZE; // Fallback in case of odd alignment + + // Calculate total chunks and generate unique message ID + const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); const messageId = Date.now().toString(); + // Send each chunk as a string with delay to manage network flow for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE); + const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, @@ -157,7 +174,9 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { const chunkWithHeader = `${chunkHeader}|${chunk}`; + // Write the chunk string directly to the socket await this.writeToSocket(chunkWithHeader); + await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay } } diff --git a/User/src/network/socket_communicator/tcp_client_communicator.ts b/User/src/network/socket_communicator/tcp_client_communicator.ts index 235d569..15d7ef4 100644 --- a/User/src/network/socket_communicator/tcp_client_communicator.ts +++ b/User/src/network/socket_communicator/tcp_client_communicator.ts @@ -5,7 +5,7 @@ import { SocketCommunicatorBase } from './socket_communicator_base'; import { OperationHandler } from '../operations_base/operation_handler'; import { operationCodes } from '../operation_codes'; -const CHUNK_SIZE = 1024; // Define chunk size +const MAX_CHUNK_SIZE = 2048; export class TcpClientCommunicator extends SocketCommunicatorBase { private readonly socket: Socket; @@ -77,11 +77,21 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); const outgoingMessage = this.encryptWithAes(message); - const totalChunks = Math.ceil(outgoingMessage.length / CHUNK_SIZE); + // Determine optimal chunk size (multiple of 4 and ≤ 1024 to align with base64 encoding) + let optimalChunkSize = MAX_CHUNK_SIZE; + while (outgoingMessage.length % optimalChunkSize !== 0 && optimalChunkSize > 0) { + optimalChunkSize -= 4; + } + + if (optimalChunkSize === 0) optimalChunkSize = MAX_CHUNK_SIZE; // Fallback in case of odd alignment + + // Calculate total chunks and generate unique message ID + const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); const messageId = Date.now().toString(); + // Send each chunk as a string with delay to manage network flow for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE); + const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, @@ -90,6 +100,7 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { const chunkWithHeader = `${chunkHeader}|${chunk}`; + // Write the chunk string directly to the socket await this.writeToSocket(chunkWithHeader); await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay } @@ -144,11 +155,13 @@ export class TcpClientCommunicator extends SocketCommunicatorBase { } async handleIncomingMessage(incomingMessage: string): Promise { - let messageToProcess = incomingMessage; + let messageToProcess; if (this.aesKey && this.aesIv) { messageToProcess = this.decryptWithAes(incomingMessage); } else if (this.serverPublicKey) { messageToProcess = this.decryptWithRsa(incomingMessage); + }else{ + messageToProcess = Buffer.from(incomingMessage, 'base64').toString('utf-8'); } const result = await this.operationHandler.handleOperation(messageToProcess); diff --git a/User/src/network/socket_communicator/tcp_server_communicator.ts b/User/src/network/socket_communicator/tcp_server_communicator.ts index 63ea4b0..fe60b59 100644 --- a/User/src/network/socket_communicator/tcp_server_communicator.ts +++ b/User/src/network/socket_communicator/tcp_server_communicator.ts @@ -4,7 +4,7 @@ import { MessageHandler } from '../message_handler'; import { SocketCommunicatorBase } from './socket_communicator_base'; import { OperationHandler } from '../operations_base/operation_handler'; -const CHUNK_SIZE = 1024; // Define chunk size +const MAX_CHUNK_SIZE = 2048; // Define chunk size export class TcpServerCommunicator extends SocketCommunicatorBase { private readonly socket: Socket; @@ -131,13 +131,18 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { } // Send chunked message - async sendChunkedMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise { + async sendChunkedMessage( + operationCode: string, + metaInfo?: { [key: string]: any }, + fileContent?: Buffer + ): Promise { const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent); let outgoingMessage: string; + // Encrypt or format message based on the operation code switch(operationCode) { case 'SET_PUBLIC_KEY': - outgoingMessage = message; + outgoingMessage = Buffer.from(message, 'utf-8').toString('base64'); break; case 'SET_AES_KEY': outgoingMessage = this.encryptWithRsa(message); @@ -146,11 +151,21 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { outgoingMessage = this.encryptWithAes(message); } - const totalChunks = Math.ceil(outgoingMessage.length / CHUNK_SIZE); + // Determine optimal chunk size (multiple of 4 and ≤ 1024 to align with base64 encoding) + let optimalChunkSize = MAX_CHUNK_SIZE; + while (outgoingMessage.length % optimalChunkSize !== 0 && optimalChunkSize > 0) { + optimalChunkSize -= 4; + } + + if (optimalChunkSize === 0) optimalChunkSize = MAX_CHUNK_SIZE; // Fallback in case of odd alignment + + // Calculate total chunks and generate unique message ID + const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize); const messageId = Date.now().toString(); + // Send each chunk as a string with delay to manage network flow for (let i = 0; i < totalChunks; i++) { - const chunk = outgoingMessage.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE); + const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize); const chunkHeader = JSON.stringify({ messageId, sequenceNumber: i + 1, @@ -159,6 +174,7 @@ export class TcpServerCommunicator extends SocketCommunicatorBase { const chunkWithHeader = `${chunkHeader}|${chunk}`; + // Write the chunk string directly to the socket await this.writeToSocket(chunkWithHeader); await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay }