admin/FACULTATE-LICENTA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

network chunk v8

Browse Source
This commit is contained in:
andrei-mihnea-cerbu
2024-11-13 10:26:53 +02:00
parent 97cc54066d
commit b2296396f5
4 changed files with 72 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
User/src/network/socket_communicator/socket_communicator_base.ts
+35
View File
@@ -1,11 +1,13 @@
import { ParsedMessage } from '../message_handler';
import { OperationHandler } from '../operations_base/operation_handler';
import { exec } from 'child_process';
export abstract class SocketCommunicatorBase {
protected readonly ip: string;
protected readonly port: number;
protected readonly operationHandler: OperationHandler;
protected handlerResult: ParsedMessage | null;
protected networkSpeed: number | null = null; // Estimated network speed in bytes/ms
protected constructor(ip: string, port: number, operationHandler: OperationHandler) {
this.ip = ip;
@@ -18,4 +20,37 @@ export abstract class SocketCommunicatorBase {
getHandlerResult(): ParsedMessage | null {
return this.handlerResult;
}
protected async scanNetworkLatency(): Promise<number> {
const targetIp = this.ip; // Use the IP from the superclass
return new Promise((resolve, reject) => {
exec(`ping -c 1 ${targetIp}`, (error, stdout) => {
if (error) {
console.error(`Ping error: ${error}`);
return reject(error);
}
const match = stdout.match(/time=([\d.]+) ms/);
if (match && match[1]) {
const latency = parseFloat(match[1]);
console.log(`Network latency to ${targetIp} is approximately ${latency} ms`);
resolve(latency);
} else {
reject(new Error('Unable to determine latency from ping output.'));
}
});
});
}
protected async calculateOptimalChunkSize(messageLength: number): Promise<number> {
this.networkSpeed = await this.scanNetworkLatency();
let chunkSize = Math.min(Math.max(512, Math.floor(5000 / this.networkSpeed)), 1024);
// Ensure chunk size aligns with base64 encoding (multiple of 4)
while (messageLength % chunkSize !== 0 && chunkSize > 0) {
chunkSize -= 4;
}
return chunkSize || 1024;
}
}
User/src/network/socket_communicator/tcp_client_communicator.ts
+9 -25
View File
@@ -12,7 +12,6 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
private serverPublicKey: string | null;
private isAesKeySetFlag: boolean;
private chunkBuffers: { [messageId: string]: string[] };
private networkSpeed: number | null = null;
constructor(socket: Socket, ip: string, port: number, operationHandler: OperationHandler) {
super(ip, port, operationHandler);
@@ -72,36 +71,20 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
}
}
private async scanNetworkSpeed(): Promise<number> {
const testMessage = 'PING_TEST'.repeat(100);
const startTime = Date.now();
await this.writeToSocket(testMessage);
await new Promise((resolve) => this.socket.once('data', resolve));
const endTime = Date.now();
const duration = endTime - startTime;
this.networkSpeed = testMessage.length / duration;
return this.networkSpeed;
}
private calculateOptimalChunkSize(messageLength: number): number {
let chunkSize = this.networkSpeed ? Math.min(Math.floor(this.networkSpeed * 100), 1024) : 1024;
while (messageLength % chunkSize !== 0 && chunkSize > 0) {
chunkSize -= 4;
}
return chunkSize || 1024;
}
async sendChunkedMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
if (!this.networkSpeed) {
await this.scanNetworkSpeed();
this.networkSpeed = await this.scanNetworkLatency();
}
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
const outgoingMessage = this.encryptWithAes(message);
const optimalChunkSize = this.calculateOptimalChunkSize(outgoingMessage.length);
// Calculate optimal chunk size based on network latency
const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length);
const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize);
const messageId = Date.now().toString();
// Send each chunk with a delay between them
for (let i = 0; i < totalChunks; i++) {
const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize);
const chunkHeader = JSON.stringify({
@@ -109,11 +92,12 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
sequenceNumber: i + 1,
totalChunks,
});
const chunkWithHeader = `${chunkHeader}|${chunk}`;
await this.writeToSocket(chunkWithHeader);
await new Promise((resolve) => setTimeout(resolve, 300));
// Set delay based on latency for smoother transmission
const delay = Math.max(100, Math.min(300, this.networkSpeed * 10));
await new Promise((resolve) => setTimeout(resolve, delay));
}
}
User/src/network/socket_communicator/tcp_server_communicator.ts
+14 -24
View File
@@ -11,7 +11,6 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
private aesKey: Buffer | null;
private aesIv: Buffer | null;
private chunkBuffers: { [messageId: string]: string[] };
private networkSpeed: number | null = null; // Estimated network speed in bytes/ms
constructor(socket: Socket, ip: string, port: number, operationHandler: OperationHandler) {
super(ip, port, operationHandler);
@@ -102,33 +101,17 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
}
}
private async scanNetworkSpeed(): Promise<number> {
const testMessage = 'PING_TEST'.repeat(100); // A test message of known size
const startTime = Date.now();
await this.writeToSocket(testMessage);
await new Promise((resolve) => this.socket.once('data', resolve));
const endTime = Date.now();
const duration = endTime - startTime; // Time in ms
this.networkSpeed = testMessage.length / duration; // Bytes/ms
return this.networkSpeed;
}
private calculateOptimalChunkSize(messageLength: number): number {
let chunkSize = this.networkSpeed ? Math.min(Math.floor(this.networkSpeed * 100), 1024) : 1024;
while (messageLength % chunkSize !== 0 && chunkSize > 0) {
chunkSize -= 4;
}
return chunkSize || 1024;
}
async sendChunkedMessage(
operationCode: string,
metaInfo?: { [key: string]: any },
fileContent?: Buffer
): Promise<void> {
// Check if latency-based chunk size needs calculation
if (!this.networkSpeed) {
await this.scanNetworkSpeed();
this.networkSpeed = await this.scanNetworkLatency();
}
// Format the message for sending
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
let outgoingMessage: string;
switch (operationCode) {
@@ -141,9 +124,13 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
default:
outgoingMessage = this.encryptWithAes(message);
}
const optimalChunkSize = this.calculateOptimalChunkSize(outgoingMessage.length);
// Calculate optimal chunk size based on network latency
const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length);
const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize);
const messageId = Date.now().toString();
// Send each chunk with a delay between them
for (let i = 0; i < totalChunks; i++) {
const chunk = outgoingMessage.slice(i * optimalChunkSize, (i + 1) * optimalChunkSize);
const chunkHeader = JSON.stringify({
@@ -153,7 +140,10 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
});
const chunkWithHeader = `${chunkHeader}|${chunk}`;
await this.writeToSocket(chunkWithHeader);
await new Promise((resolve) => setTimeout(resolve, 300));
// Set delay based on latency for smoother transmission
const delay = Math.max(100, Math.min(300, this.networkSpeed * 10));
await new Promise((resolve) => setTimeout(resolve, delay));
}
}
@@ -165,7 +155,7 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
this.handlerResult = await this.operationHandler.handleOperation(messageToProcess);
}
private writeToSocket(message: string): Promise<void> {
protected writeToSocket(message: string): Promise<void> {
return new Promise((resolve, reject) => {
this.socket.write(message, (err: any) => {
if (err) {