network chunk v6
This commit is contained in:
@@ -4,7 +4,7 @@ import { MessageHandler } from '../message_handler';
|
|||||||
import { SocketCommunicatorBase } from './socket_communicator_base';
|
import { SocketCommunicatorBase } from './socket_communicator_base';
|
||||||
import { OperationHandler } from '../operations_base/operation_handler';
|
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 {
|
export class TcpServerCommunicator extends SocketCommunicatorBase {
|
||||||
private readonly socket: Socket;
|
private readonly socket: Socket;
|
||||||
@@ -100,6 +100,8 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
const [headerJson, chunkContent] = incomingMessage.split('|');
|
const [headerJson, chunkContent] = incomingMessage.split('|');
|
||||||
const header = JSON.parse(headerJson);
|
const header = JSON.parse(headerJson);
|
||||||
|
|
||||||
|
console.log(`\n\n${incomingMessage}\n\n`);
|
||||||
|
|
||||||
// Initialize chunk array if this is the first chunk for this messageId
|
// Initialize chunk array if this is the first chunk for this messageId
|
||||||
if (!this.chunkBuffers[header.messageId]) {
|
if (!this.chunkBuffers[header.messageId]) {
|
||||||
this.chunkBuffers[header.messageId] = new Array(header.totalChunks);
|
this.chunkBuffers[header.messageId] = new Array(header.totalChunks);
|
||||||
@@ -129,13 +131,18 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Send chunked message
|
// Send chunked message
|
||||||
async sendChunkedMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
|
async sendChunkedMessage(
|
||||||
|
operationCode: string,
|
||||||
|
metaInfo?: { [key: string]: any },
|
||||||
|
fileContent?: Buffer
|
||||||
|
): Promise<void> {
|
||||||
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
|
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
|
||||||
let outgoingMessage: string;
|
let outgoingMessage: string;
|
||||||
|
|
||||||
|
// Encrypt or format message based on the operation code
|
||||||
switch(operationCode) {
|
switch(operationCode) {
|
||||||
case 'SET_PUBLIC_KEY':
|
case 'SET_PUBLIC_KEY':
|
||||||
outgoingMessage = message;
|
outgoingMessage = Buffer.from(message, 'utf-8').toString('base64');
|
||||||
break;
|
break;
|
||||||
case 'SET_AES_KEY':
|
case 'SET_AES_KEY':
|
||||||
outgoingMessage = this.encryptWithRsa(message);
|
outgoingMessage = this.encryptWithRsa(message);
|
||||||
@@ -144,11 +151,21 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
outgoingMessage = this.encryptWithAes(message);
|
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();
|
const messageId = Date.now().toString();
|
||||||
|
|
||||||
|
// Send each chunk as a string with delay to manage network flow
|
||||||
for (let i = 0; i < totalChunks; i++) {
|
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({
|
const chunkHeader = JSON.stringify({
|
||||||
messageId,
|
messageId,
|
||||||
sequenceNumber: i + 1,
|
sequenceNumber: i + 1,
|
||||||
@@ -157,7 +174,9 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
|
|
||||||
const chunkWithHeader = `${chunkHeader}|${chunk}`;
|
const chunkWithHeader = `${chunkHeader}|${chunk}`;
|
||||||
|
|
||||||
|
// Write the chunk string directly to the socket
|
||||||
await this.writeToSocket(chunkWithHeader);
|
await this.writeToSocket(chunkWithHeader);
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import { SocketCommunicatorBase } from './socket_communicator_base';
|
|||||||
import { OperationHandler } from '../operations_base/operation_handler';
|
import { OperationHandler } from '../operations_base/operation_handler';
|
||||||
import { operationCodes } from '../operation_codes';
|
import { operationCodes } from '../operation_codes';
|
||||||
|
|
||||||
const CHUNK_SIZE = 1024; // Define chunk size
|
const MAX_CHUNK_SIZE = 2048;
|
||||||
|
|
||||||
export class TcpClientCommunicator extends SocketCommunicatorBase {
|
export class TcpClientCommunicator extends SocketCommunicatorBase {
|
||||||
private readonly socket: Socket;
|
private readonly socket: Socket;
|
||||||
@@ -77,11 +77,21 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
|
|||||||
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
|
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
|
||||||
const outgoingMessage = this.encryptWithAes(message);
|
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();
|
const messageId = Date.now().toString();
|
||||||
|
|
||||||
|
// Send each chunk as a string with delay to manage network flow
|
||||||
for (let i = 0; i < totalChunks; i++) {
|
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({
|
const chunkHeader = JSON.stringify({
|
||||||
messageId,
|
messageId,
|
||||||
sequenceNumber: i + 1,
|
sequenceNumber: i + 1,
|
||||||
@@ -90,6 +100,7 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
|
|||||||
|
|
||||||
const chunkWithHeader = `${chunkHeader}|${chunk}`;
|
const chunkWithHeader = `${chunkHeader}|${chunk}`;
|
||||||
|
|
||||||
|
// Write the chunk string directly to the socket
|
||||||
await this.writeToSocket(chunkWithHeader);
|
await this.writeToSocket(chunkWithHeader);
|
||||||
await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay
|
await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay
|
||||||
}
|
}
|
||||||
@@ -144,11 +155,13 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async handleIncomingMessage(incomingMessage: string): Promise<void> {
|
async handleIncomingMessage(incomingMessage: string): Promise<void> {
|
||||||
let messageToProcess = incomingMessage;
|
let messageToProcess;
|
||||||
if (this.aesKey && this.aesIv) {
|
if (this.aesKey && this.aesIv) {
|
||||||
messageToProcess = this.decryptWithAes(incomingMessage);
|
messageToProcess = this.decryptWithAes(incomingMessage);
|
||||||
} else if (this.serverPublicKey) {
|
} else if (this.serverPublicKey) {
|
||||||
messageToProcess = this.decryptWithRsa(incomingMessage);
|
messageToProcess = this.decryptWithRsa(incomingMessage);
|
||||||
|
}else{
|
||||||
|
messageToProcess = Buffer.from(incomingMessage, 'base64').toString('utf-8');
|
||||||
}
|
}
|
||||||
|
|
||||||
const result = await this.operationHandler.handleOperation(messageToProcess);
|
const result = await this.operationHandler.handleOperation(messageToProcess);
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import { MessageHandler } from '../message_handler';
|
|||||||
import { SocketCommunicatorBase } from './socket_communicator_base';
|
import { SocketCommunicatorBase } from './socket_communicator_base';
|
||||||
import { OperationHandler } from '../operations_base/operation_handler';
|
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 {
|
export class TcpServerCommunicator extends SocketCommunicatorBase {
|
||||||
private readonly socket: Socket;
|
private readonly socket: Socket;
|
||||||
@@ -131,13 +131,18 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Send chunked message
|
// Send chunked message
|
||||||
async sendChunkedMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
|
async sendChunkedMessage(
|
||||||
|
operationCode: string,
|
||||||
|
metaInfo?: { [key: string]: any },
|
||||||
|
fileContent?: Buffer
|
||||||
|
): Promise<void> {
|
||||||
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
|
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
|
||||||
let outgoingMessage: string;
|
let outgoingMessage: string;
|
||||||
|
|
||||||
|
// Encrypt or format message based on the operation code
|
||||||
switch(operationCode) {
|
switch(operationCode) {
|
||||||
case 'SET_PUBLIC_KEY':
|
case 'SET_PUBLIC_KEY':
|
||||||
outgoingMessage = message;
|
outgoingMessage = Buffer.from(message, 'utf-8').toString('base64');
|
||||||
break;
|
break;
|
||||||
case 'SET_AES_KEY':
|
case 'SET_AES_KEY':
|
||||||
outgoingMessage = this.encryptWithRsa(message);
|
outgoingMessage = this.encryptWithRsa(message);
|
||||||
@@ -146,11 +151,21 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
outgoingMessage = this.encryptWithAes(message);
|
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();
|
const messageId = Date.now().toString();
|
||||||
|
|
||||||
|
// Send each chunk as a string with delay to manage network flow
|
||||||
for (let i = 0; i < totalChunks; i++) {
|
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({
|
const chunkHeader = JSON.stringify({
|
||||||
messageId,
|
messageId,
|
||||||
sequenceNumber: i + 1,
|
sequenceNumber: i + 1,
|
||||||
@@ -159,6 +174,7 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
|
|||||||
|
|
||||||
const chunkWithHeader = `${chunkHeader}|${chunk}`;
|
const chunkWithHeader = `${chunkHeader}|${chunk}`;
|
||||||
|
|
||||||
|
// Write the chunk string directly to the socket
|
||||||
await this.writeToSocket(chunkWithHeader);
|
await this.writeToSocket(chunkWithHeader);
|
||||||
await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay
|
await new Promise((resolve) => setTimeout(resolve, 300)); // Simulate network delay
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user