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

network chunk v20

Browse Source
This commit is contained in:
andrei-mihnea-cerbu
2024-11-13 17:23:21 +02:00
parent f95b67f93d
commit 9baec7a7cb
41 changed files with 714 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CEO/src/network/operation_codes.ts
+2 -1
View File
@@ -1,6 +1,7 @@
export let operationCodes = {
// General Operations
HEARTBEAT: 'HEARTBEAT',
ARE_YOU_HUMAN: 'ARE_YOU_HUMAN',
ARE_YOU_UC: 'ARE_YOU_UC',
ALIVE: 'ALIVE',
SET_PUBLIC_KEY: 'SET_PUBLIC_KEY',
SET_AES_KEY: 'SET_AES_KEY',
CEO/src/network/operations_custom/general_operations.ts
+3 -3
View File
@@ -7,14 +7,14 @@ export class GeneralOperations implements OperationPlugin {
public static readonly operationCodes = {
OK: 'OK',
ERR: 'ERR',
HEARTBEAT: 'HEARTBEAT',
ARE_YOU_HUMAN: 'ARE_YOU_HUMAN',
ALIVE: 'ALIVE',
SET_PUBLIC_KEY: 'SET_PUBLIC_KEY',
SET_AES_KEY: 'SET_AES_KEY',
};
// Handle heartbeat operation asynchronously
public static async handleHeartbeat(): Promise<ParsedMessage> {
public static async handleAreYouHuman(): Promise<ParsedMessage> {
const networkInterfaces = os.networkInterfaces();
let ipAddress = 'Unknown';
@@ -79,7 +79,7 @@ export class GeneralOperations implements OperationPlugin {
// Register general operations with the OperationHandler
public register(operationHandler: OperationHandler): void {
operationHandler.registerHandler(GeneralOperations.operationCodes.HEARTBEAT, GeneralOperations.handleHeartbeat);
operationHandler.registerHandler(GeneralOperations.operationCodes.ARE_YOU_HUMAN, GeneralOperations.handleAreYouHuman);
operationHandler.registerHandler(GeneralOperations.operationCodes.SET_PUBLIC_KEY, GeneralOperations.handlePublicKey);
operationHandler.registerHandler(GeneralOperations.operationCodes.SET_AES_KEY, GeneralOperations.handleAESKey);
operationHandler.registerHandler(GeneralOperations.operationCodes.OK, GeneralOperations.handleOk);
CEO/src/network/socket_communicator/socket_communicator_base.ts
+2 -40
View File
@@ -1,6 +1,5 @@
import { ParsedMessage } from '../message_handler';
import { OperationHandler } from '../operations_base/operation_handler';
import ping from "ping";
import {
constants,
createCipheriv,
@@ -16,16 +15,16 @@ export abstract class SocketCommunicatorBase {
protected readonly port: number;
protected readonly operationHandler: OperationHandler;
protected handlerResult: ParsedMessage | null;
protected networkSpeed: number | null = null;
protected chunkBuffers: { [messageId: string]: string[] };
protected readonly EOP = '<EOP>';
protected privateKey: string | null;
protected publicKey: string | null;
protected aesKey: Buffer | null;
protected aesIv: Buffer | null;
protected readonly EOP = '<EOP>';
protected readonly CHUNK_SIZE = 1024;
private incompleteChunkBuffer: string = '';
protected constructor(ip: string, port: number, operationHandler: OperationHandler) {
@@ -113,40 +112,6 @@ export abstract class SocketCommunicatorBase {
).toString('base64');
}
protected async scanNetworkLatency(): Promise<number> {
const targetIp = this.ip; // Use the IP from the superclass
try {
const response = await ping.promise.probe(targetIp);
if (!response.alive || response.time === "unknown") {
console.warn(`Ping failed to reach ${targetIp}. Using default network speed.`);
return 200; // Default latency in ms if ping fails
}
return response.time; // Latency in ms from ping response
} catch (error: any) {
console.error(`Ping error: ${error.message}. Using default network speed.`);
return 200; // Default latency in ms if an error occurs
}
}
// Calculate optimal chunk size based on network latency, with fallback if necessary
protected async calculateOptimalChunkSize(messageLength: number): Promise<number> {
const latency = await this.scanNetworkLatency();
this.networkSpeed = latency > 0 ? 1000 / latency : 1; // Speed in bytes/ms based on latency
// Calculate initial chunk size based on latency (bounded between 512 and 1024 bytes)
let chunkSize = Math.min(Math.max(512, Math.floor(5000 / this.networkSpeed)), 1024);
// Adjust chunk size for base64 alignment (multiple of 4)
while (messageLength % chunkSize !== 0 && chunkSize > 0) {
chunkSize -= 4;
}
return chunkSize || 1024; // Fallback to 1024 if alignment adjustment results in 0
}
async handleIncomingChunk(data: Buffer): Promise<void> {
// Append incoming data to the incomplete buffer
this.incompleteChunkBuffer += data.toString();
@@ -171,9 +136,6 @@ export abstract class SocketCommunicatorBase {
// Store the chunk in the correct position based on sequenceNumber (1-based indexing)
this.chunkBuffers[header.messageId][header.sequenceNumber - 1] = chunkContent;
console.log(`Received chunk: ${incomingMessage}`);
console.log(`Chunks received so far: ${this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length}`);
// Check if all chunks have been received
if (this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length === header.totalChunks) {
// Join all chunks to form the full message
CEO/src/network/socket_communicator/tcp_client_communicator.ts
+8 -10
View File
@@ -22,12 +22,10 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
}
setServerPublicKey(publicKey: string): void {
console.log('\n\nSetting server public key\n\n');
this.publicKey = publicKey;
}
setAesKey(aesKey: string, aesIv: string): void {
console.log('\n\nSetting AES key\n\n');
this.aesKey = Buffer.from(aesKey, 'base64');
this.aesIv = Buffer.from(aesIv, 'base64');
}
@@ -60,29 +58,29 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
}
async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
if (!this.networkSpeed) {
this.networkSpeed = await this.scanNetworkLatency();
}
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
const outgoingMessage = this.encryptWithAes(message);
// Calculate optimal chunk size based on network latency
const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length);
const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize);
const totalChunks = Math.ceil(outgoingMessage.length / this.CHUNK_SIZE);
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 chunk = outgoingMessage.slice(i * this.CHUNK_SIZE, (i + 1) * this.CHUNK_SIZE);
const chunkHeader = JSON.stringify({
messageId,
sequenceNumber: i + 1,
totalChunks,
});
const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`;
if(!this.socket.write(chunkWithHeader)) this.socket.end();
if (!this.socket.write(chunkWithHeader)) {
// Wait for the 'drain' event before writing the next chunk
await new Promise((resolve) => this.socket.once('drain', resolve));
}
}
}
CEO/src/network/socket_communicator/tcp_server_communicator.ts
+7 -9
View File
@@ -38,10 +38,6 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
}
async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
if (!this.networkSpeed) {
this.networkSpeed = await this.scanNetworkLatency();
}
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
let outgoingMessage: string;
@@ -56,21 +52,23 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
outgoingMessage = this.encryptWithAes(message);
}
// Calculate optimal chunk size based on network latency
const optimalChunkSize = await this.calculateOptimalChunkSize(outgoingMessage.length);
const totalChunks = Math.ceil(outgoingMessage.length / optimalChunkSize);
const totalChunks = Math.ceil(outgoingMessage.length / this.CHUNK_SIZE);
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 chunk = outgoingMessage.slice(i * this.CHUNK_SIZE, (i + 1) * this.CHUNK_SIZE);
const chunkHeader = JSON.stringify({
messageId,
sequenceNumber: i + 1,
totalChunks,
});
const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`;
if(!this.socket.write(chunkWithHeader)) this.socket.end();
if (!this.socket.write(chunkWithHeader)) {
// Wait for the 'drain' event before writing the next chunk
await new Promise((resolve) => this.socket.once('drain', resolve));
}
}
}
}
CEO/src/network/udp/udp_client.ts
+3 -4
View File
@@ -29,7 +29,7 @@ export class UdpClient {
}
// Send heartbeat and get clients that respond with ALIVE, excluding the host machine's IP addresses
async getAliveClients(): Promise<string[]> {
async getTargetClients(heartbeatCode: string): Promise<string[]> {
const subnet = this.getSubnet();
const ipRange = this.getIPRange(subnet);
@@ -45,7 +45,7 @@ export class UdpClient {
const aliveClients: string[] = [];
for (const ip of activeIps) {
if (!localIPs.includes(ip)) {
const result = await this.sendHeartbeat(ip);
const result = await this.sendHeartbeat(ip, heartbeatCode);
if (result.found) {
aliveClients.push(ip);
}
@@ -73,9 +73,8 @@ export class UdpClient {
}
// Send heartbeat to an IP
private async sendHeartbeat(ip: string): Promise<{ found: boolean }> {
private async sendHeartbeat(ip: string, heartbeatCode: string): Promise<{ found: boolean }> {
return new Promise((resolve) => {
const heartbeatCode = operationCodes.HEARTBEAT;
const heartbeatMessage = MessageHandler.formatMessage(heartbeatCode);
this.log(`Sending heartbeat to ${ip}`);