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

network chunk v2

Browse Source
This commit is contained in:
andrei-mihnea-cerbu
2024-11-12 23:14:08 +02:00
parent 44df437425
commit fa4750ea3b
4 changed files with 53 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
User/src/helpers/file_sharer.ts
+3 -9
View File
@@ -28,24 +28,18 @@ export class FileSharer {
async start(): Promise<void> { async start(): Promise<void> {
setInterval(async () => { setInterval(async () => {
if (!this.isBusy) { // Check if the queue is already being processed if (!this.isBusy) { // Check if the queue is already being processed
this.isBusy = true; // Set busy flag to true before starting
this.log("Start successfully. Processing the queue."); this.log("Start successfully. Processing the queue.");
await this.processQueue(); // Process the queue at regular intervals await this.processQueue();
this.log("Queue processing completed.");
} }
}, 10000); // 10 seconds interval }, 10000); // 10 seconds interval
} }
// Method to process the queue // Method to process the queue
private async processQueue(): Promise<void> { private async processQueue(): Promise<void> {
if (this.isBusy) {
this.log("Queue is already being processed. Skipping this interval.");
return;
}
this.isBusy = true; // Set busy flag to true before starting
while (!this.queueManager.isEmpty()) { while (!this.queueManager.isEmpty()) {
const task = this.queueManager.peek(); const task = this.queueManager.peek();
this.log('trimiti fisier');
if (task) { if (task) {
this.log(`Processing task for file: ${task.path} to IP: ${task.ip}`); this.log(`Processing task for file: ${task.path} to IP: ${task.ip}`);
User/src/network/operations_custom/user_to_user_operations.ts
+2
View File
@@ -116,6 +116,8 @@ export class UserToUserOperations implements OperationPlugin {
const appInfo = await jsonManager.readValue('shareDirectory'); const appInfo = await jsonManager.readValue('shareDirectory');
const shareDirectory = appInfo?.path || ''; const shareDirectory = appInfo?.path || '';
console.log(`\n\nShare directory: ${shareDirectory}\n\n`);
if (!shareDirectory) { if (!shareDirectory) {
return { operationCode: operationCodes.ERR, metaInfo: { message: 'Share directory missing.' }}; return { operationCode: operationCodes.ERR, metaInfo: { message: 'Share directory missing.' }};
} }
User/src/network/socket_communicator/tcp_client_communicator.ts
+24 -20
View File
@@ -113,37 +113,41 @@ export class TcpClientCommunicator extends SocketCommunicatorBase {
} }
async handleIncomingChunk(data: Buffer): Promise<void> { async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingMessage = data.toString(); const incomingMessage = data.toString().trim();
this.messageBuffer += incomingMessage;
if (this.messageBuffer.includes(END_OF_MESSAGE)) { // Check if the message is the end marker <EOM>
const messages = this.messageBuffer.split(END_OF_MESSAGE); if (incomingMessage === END_OF_MESSAGE) {
// Process the complete message when EOM is received
for (let i = 0; i < messages.length - 1; i++) { await this.processCompleteMessage();
const completeMessage = messages[i]; return;
if (completeMessage) {
await this.processCompleteMessage(completeMessage);
}
}
this.messageBuffer = messages[messages.length - 1];
} }
}
private async processCompleteMessage(completeMessage: string): Promise<void> { // Extract header and chunk content from incoming data
const [headerJson, chunkContent] = completeMessage.split('|'); const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson); const header = JSON.parse(headerJson);
// 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);
} }
// Place the chunk in the correct position in the chunk buffer
this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent; this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent;
}
if (this.chunkBuffers[header.messageId].every((chunk) => chunk !== undefined)) { // Process the complete message when EOM is received
const fullMessage = this.chunkBuffers[header.messageId].join(''); private async processCompleteMessage(): Promise<void> {
await this.handleIncomingMessage(fullMessage); for (const [messageId, chunks] of Object.entries(this.chunkBuffers)) {
delete this.chunkBuffers[header.messageId]; if (chunks.every((chunk) => chunk !== undefined)) {
// Join all chunks to form the full message
const fullMessage = chunks.join('');
// Handle the completed and possibly decrypted message
await this.handleIncomingMessage(fullMessage);
// Clean up buffer after processing
delete this.chunkBuffers[messageId];
}
} }
} }
User/src/network/socket_communicator/tcp_server_communicator.ts
+24 -20
View File
@@ -97,37 +97,41 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
// Handle incoming chunks of data // Handle incoming chunks of data
async handleIncomingChunk(data: Buffer): Promise<void> { async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingMessage = data.toString(); const incomingMessage = data.toString().trim();
this.messageBuffer += incomingMessage;
if (this.messageBuffer.includes(END_OF_MESSAGE)) { // Check if the message is the end marker <EOM>
const messages = this.messageBuffer.split(END_OF_MESSAGE); if (incomingMessage === END_OF_MESSAGE) {
// Process the complete message when EOM is received
for (let i = 0; i < messages.length - 1; i++) { await this.processCompleteMessage();
const completeMessage = messages[i]; return;
if (completeMessage) {
await this.processCompleteMessage(completeMessage);
}
}
this.messageBuffer = messages[messages.length - 1];
} }
}
private async processCompleteMessage(completeMessage: string): Promise<void> { // Extract header and chunk content from incoming data
const [headerJson, chunkContent] = completeMessage.split('|'); const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson); const header = JSON.parse(headerJson);
// 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);
} }
// Place the chunk in the correct position in the chunk buffer
this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent; this.chunkBuffers[header.messageId][header.sequenceNumber] = chunkContent;
}
if (this.chunkBuffers[header.messageId].every((chunk) => chunk !== undefined)) { // Process the complete message when EOM is received
const fullMessage = this.chunkBuffers[header.messageId].join(''); private async processCompleteMessage(): Promise<void> {
await this.handleIncomingMessage(fullMessage); for (const [messageId, chunks] of Object.entries(this.chunkBuffers)) {
delete this.chunkBuffers[header.messageId]; if (chunks.every((chunk) => chunk !== undefined)) {
// Join all chunks to form the full message
const fullMessage = chunks.join('');
// Handle the completed and possibly decrypted message
await this.handleIncomingMessage(fullMessage);
// Clean up buffer after processing
delete this.chunkBuffers[messageId];
}
} }
} }