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

network chunk v14

Browse Source
This commit is contained in:
andrei-mihnea-cerbu
2024-11-13 14:14:38 +02:00
parent 116676db77
commit fa974d4d9f
10 changed files with 319 additions and 335 deletions
Download Patch File Download Diff File Expand all files Collapse all files
UC/src/network/socket_communicator/socket_communicator_base.ts
+32 -19
View File
@@ -26,6 +26,8 @@ export abstract class SocketCommunicatorBase {
protected aesKey: Buffer | null;
protected aesIv: Buffer | null;
private incompleteChunkBuffer: string = '';
protected constructor(ip: string, port: number, operationHandler: OperationHandler) {
this.ip = ip;
this.port = port;
@@ -146,34 +148,45 @@ export abstract class SocketCommunicatorBase {
}
async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingData = data.toString().trim();
// Append incoming data to the incomplete buffer
this.incompleteChunkBuffer += data.toString();
// Split the incoming data by <EOP> to handle multiple chunks concatenated by TCP
const messages = incomingData.split(this.EOP).filter(Boolean); // Filter out any empty strings from split
// Split the buffer by <EOP> to separate complete and incomplete messages
const messages = this.incompleteChunkBuffer.split(this.EOP);
// Save the last item back to the buffer if it's incomplete (no <EOP> at the end)
this.incompleteChunkBuffer = messages.pop() || "";
// Process each complete message in the split results
for (const incomingMessage of messages) {
const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson);
try {
const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson);
// Initialize an array for chunks if it's the first chunk received for this messageId
if (!this.chunkBuffers[header.messageId]) {
this.chunkBuffers[header.messageId] = [];
}
// Initialize an array for chunks if it's the first chunk for this messageId
if (!this.chunkBuffers[header.messageId]) {
this.chunkBuffers[header.messageId] = [];
}
// Directly set the chunk at the correct index, adjusting for 1-based indexing
this.chunkBuffers[header.messageId][header.sequenceNumber - 1] = chunkContent;
// 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}`);
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 by confirming the length
if (this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length === header.totalChunks) {
const chunks = this.chunkBuffers[header.messageId];
const fullMessage = chunks.join('');
// 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
const fullMessage = this.chunkBuffers[header.messageId].join('');
await this.handleIncomingMessage(fullMessage);
// Process the complete message
await this.handleIncomingMessage(fullMessage);
delete this.chunkBuffers[header.messageId];
// Clear the chunk buffer for this messageId
delete this.chunkBuffers[header.messageId];
}
} catch (error: any) {
console.error(`Error handling chunk: ${error.message}`);
}
}
}