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

network chunk v3

Browse Source
This commit is contained in:
andrei-mihnea-cerbu
2024-11-12 23:28:12 +02:00
parent fa4750ea3b
commit 8584a0683f
2 changed files with 9 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
User/src/network/socket_communicator/tcp_client_communicator.ts
+5 -12
View File
@@ -5,7 +5,6 @@ import { SocketCommunicatorBase } from './socket_communicator_base';
import { OperationHandler } from '../operations_base/operation_handler';
import { operationCodes } from '../operation_codes';
const END_OF_MESSAGE = '<EOM>'; // Unique marker for the end of message
const CHUNK_SIZE = 1024; // Define chunk size
export class TcpClientCommunicator extends SocketCommunicatorBase {
@@ -87,17 +86,13 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
const chunk = outgoingMessage.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE);
const chunkHeader = JSON.stringify({
messageId,
sequenceNumber: i,
sequenceNumber: i + 1,
totalChunks,
});
const chunkWithHeader = `${chunkHeader}|${chunk}`;
await this.writeToSocket(chunkWithHeader);
if (i === totalChunks - 1) {
await this.writeToSocket(END_OF_MESSAGE);
}
}
}
@@ -115,12 +110,7 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingMessage = data.toString().trim();
// Check if the message is the end marker <EOM>
if (incomingMessage === END_OF_MESSAGE) {
// Process the complete message when EOM is received
await this.processCompleteMessage();
return;
}
console.log(`\n\n${incomingMessage}\n\n`);
// Extract header and chunk content from incoming data
const [headerJson, chunkContent] = incomingMessage.split('|');
@@ -133,6 +123,9 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
// Place the chunk in the correct position in the chunk buffer
this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent;
if(header.sequenceNumber === header.totalChunks) {
await this.processCompleteMessage();
}
}
// Process the complete message when EOM is received
User/src/network/socket_communicator/tcp_server_communicator.ts
+4 -13
View File
@@ -4,7 +4,6 @@ import { MessageHandler } from '../message_handler';
import { SocketCommunicatorBase } from './socket_communicator_base';
import { OperationHandler } from '../operations_base/operation_handler';
const END_OF_MESSAGE = '<EOM>'; // Define a unique marker for end of message
const CHUNK_SIZE = 1024; // Define chunk size
export class TcpServerCommunicator extends SocketCommunicatorBase {
@@ -99,13 +98,6 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingMessage = data.toString().trim();
// Check if the message is the end marker <EOM>
if (incomingMessage === END_OF_MESSAGE) {
// Process the complete message when EOM is received
await this.processCompleteMessage();
return;
}
// Extract header and chunk content from incoming data
const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson);
@@ -117,6 +109,9 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
// Place the chunk in the correct position in the chunk buffer
this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent;
if(header.sequenceNumber === header.totalChunks) {
await this.processCompleteMessage();
}
}
// Process the complete message when EOM is received
@@ -158,17 +153,13 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
const chunk = outgoingMessage.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE);
const chunkHeader = JSON.stringify({
messageId,
sequenceNumber: i,
sequenceNumber: i + 1,
totalChunks,
});
const chunkWithHeader = `${chunkHeader}|${chunk}`;
await this.writeToSocket(chunkWithHeader);
if (i === totalChunks - 1) {
await this.writeToSocket(END_OF_MESSAGE);
}
}
}