chore:Split the fileReceiver.ts
This commit is contained in:
david_bai
+141
-1344
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,343 @@
|
||||
import { EmbeddedChunkMeta } from "@/types/webrtc";
|
||||
import { ReceptionConfig } from "./ReceptionConfig";
|
||||
import { postLogToBackend } from "@/app/config/api";
|
||||
|
||||
const developmentEnv = process.env.NODE_ENV;
|
||||
|
||||
/**
|
||||
* 🚀 Chunk processing result interface
|
||||
*/
|
||||
export interface ChunkProcessingResult {
|
||||
chunkMeta: EmbeddedChunkMeta;
|
||||
chunkData: ArrayBuffer;
|
||||
absoluteChunkIndex: number;
|
||||
relativeChunkIndex: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* 🚀 Chunk processor
|
||||
* Handles all data chunk processing, format conversion, and parsing
|
||||
*/
|
||||
export class ChunkProcessor {
|
||||
/**
|
||||
* Convert various binary data formats to ArrayBuffer
|
||||
* Supports Blob, Uint8Array, and other formats for Firefox compatibility
|
||||
*/
|
||||
async convertToArrayBuffer(data: any): Promise<ArrayBuffer | null> {
|
||||
const originalType = Object.prototype.toString.call(data);
|
||||
|
||||
if (data instanceof ArrayBuffer) {
|
||||
return data;
|
||||
} else if (data instanceof Blob) {
|
||||
try {
|
||||
const arrayBuffer = await data.arrayBuffer();
|
||||
if (data.size !== arrayBuffer.byteLength) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ⚠️ Blob size mismatch: ${data.size}→${arrayBuffer.byteLength}`
|
||||
);
|
||||
}
|
||||
}
|
||||
return arrayBuffer;
|
||||
} catch (error) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(`[DEBUG] ❌ Blob conversion failed: ${error}`);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
} else if (data instanceof Uint8Array || ArrayBuffer.isView(data)) {
|
||||
try {
|
||||
const uint8Array =
|
||||
data instanceof Uint8Array
|
||||
? data
|
||||
: new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
|
||||
const newArrayBuffer = new ArrayBuffer(uint8Array.length);
|
||||
new Uint8Array(newArrayBuffer).set(uint8Array);
|
||||
return newArrayBuffer;
|
||||
} catch (error) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(`[DEBUG] ❌ TypedArray conversion failed: ${error}`);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
} else {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ❌ Unknown data type: ${Object.prototype.toString.call(
|
||||
data
|
||||
)}`
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse embedded chunk packet
|
||||
* Format: [4 bytes length] + [JSON metadata] + [actual chunk data]
|
||||
*/
|
||||
parseEmbeddedChunkPacket(arrayBuffer: ArrayBuffer): {
|
||||
chunkMeta: EmbeddedChunkMeta;
|
||||
chunkData: ArrayBuffer;
|
||||
} | null {
|
||||
try {
|
||||
// 1. Check minimum packet length
|
||||
if (arrayBuffer.byteLength < ReceptionConfig.VALIDATION_CONFIG.MIN_PACKET_SIZE) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ❌ Invalid embedded packet - too small: ${arrayBuffer.byteLength}`
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 2. Read metadata length (4 bytes)
|
||||
const lengthView = new Uint32Array(arrayBuffer, 0, 1);
|
||||
const metaLength = lengthView[0];
|
||||
|
||||
// 3. Verify packet integrity
|
||||
const expectedTotalLength = 4 + metaLength;
|
||||
if (arrayBuffer.byteLength < expectedTotalLength) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ❌ Incomplete embedded packet - expected: ${expectedTotalLength}, got: ${arrayBuffer.byteLength}`
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 4. Extract metadata section
|
||||
const metaBytes = new Uint8Array(arrayBuffer, 4, metaLength);
|
||||
const metaJson = new TextDecoder().decode(metaBytes);
|
||||
const chunkMeta: EmbeddedChunkMeta = JSON.parse(metaJson);
|
||||
|
||||
// 5. Extract actual chunk data section
|
||||
const chunkDataStart = 4 + metaLength;
|
||||
const chunkData = arrayBuffer.slice(chunkDataStart);
|
||||
|
||||
// 6. Verify chunk data size
|
||||
if (chunkData.byteLength !== chunkMeta.chunkSize) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ⚠️ Chunk size mismatch - meta: ${chunkMeta.chunkSize}, actual: ${chunkData.byteLength}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return { chunkMeta, chunkData };
|
||||
} catch (error) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ❌ Failed to parse embedded packet: ${error}`
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process received chunk and calculate indices
|
||||
*/
|
||||
processReceivedChunk(
|
||||
chunkMeta: EmbeddedChunkMeta,
|
||||
chunkData: ArrayBuffer,
|
||||
initialOffset: number
|
||||
): ChunkProcessingResult | null {
|
||||
// Calculate indices
|
||||
const absoluteChunkIndex = chunkMeta.chunkIndex; // Sender's absolute index
|
||||
const startChunkIndex = ReceptionConfig.getChunkIndexFromOffset(initialOffset); // Resume start index
|
||||
const relativeChunkIndex = absoluteChunkIndex - startChunkIndex; // Relative index in chunks array
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING && absoluteChunkIndex <= 970) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] Index mapping - absolute:${absoluteChunkIndex}, start:${startChunkIndex}, relative:${relativeChunkIndex}`
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
chunkMeta,
|
||||
chunkData,
|
||||
absoluteChunkIndex,
|
||||
relativeChunkIndex,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate chunk against expected parameters
|
||||
*/
|
||||
validateChunk(
|
||||
chunkMeta: EmbeddedChunkMeta,
|
||||
expectedFileId: string,
|
||||
expectedChunksCount: number,
|
||||
initialOffset: number
|
||||
): {
|
||||
isValid: boolean;
|
||||
errors: string[];
|
||||
} {
|
||||
const errors: string[] = [];
|
||||
|
||||
// Verify fileId match
|
||||
if (chunkMeta.fileId !== expectedFileId) {
|
||||
errors.push(`FileId mismatch - expected: ${expectedFileId}, got: ${chunkMeta.fileId}`);
|
||||
}
|
||||
|
||||
// Validate chunk size
|
||||
if (chunkMeta.chunkSize <= 0) {
|
||||
errors.push(`Invalid chunk size: ${chunkMeta.chunkSize}`);
|
||||
}
|
||||
|
||||
// Check if chunk index is reasonable
|
||||
if (chunkMeta.chunkIndex < 0) {
|
||||
errors.push(`Invalid chunk index: ${chunkMeta.chunkIndex}`);
|
||||
}
|
||||
|
||||
// Validate total chunks (with resume consideration)
|
||||
const startChunkIndex = ReceptionConfig.getChunkIndexFromOffset(initialOffset);
|
||||
const calculatedExpected = chunkMeta.totalChunks - startChunkIndex;
|
||||
|
||||
if (chunkMeta.totalChunks !== expectedChunksCount && calculatedExpected !== expectedChunksCount) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] Chunk count info - fileTotal: ${chunkMeta.totalChunks}, currentExpected: ${expectedChunksCount}, calculatedExpected: ${calculatedExpected}, startChunk: ${startChunkIndex}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
isValid: errors.length === 0,
|
||||
errors,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if chunk index is within valid range
|
||||
*/
|
||||
isChunkIndexValid(
|
||||
relativeChunkIndex: number,
|
||||
expectedChunksCount: number
|
||||
): boolean {
|
||||
return relativeChunkIndex >= 0 && relativeChunkIndex < expectedChunksCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Log chunk processing details (for debugging)
|
||||
*/
|
||||
logChunkDetails(
|
||||
result: ChunkProcessingResult,
|
||||
expectedChunksCount: number,
|
||||
writerExpectedIndex?: number
|
||||
): void {
|
||||
if (!ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
return;
|
||||
}
|
||||
|
||||
const { chunkMeta, absoluteChunkIndex, relativeChunkIndex } = result;
|
||||
const lastFewChunks = relativeChunkIndex >= expectedChunksCount - 5;
|
||||
|
||||
if (absoluteChunkIndex <= 970 || lastFewChunks) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] 📦 Chunk #${absoluteChunkIndex} received - relative:${relativeChunkIndex}, size:${chunkMeta.chunkSize}${
|
||||
writerExpectedIndex !== undefined ? `, writerExpects:${writerExpectedIndex}` : ''
|
||||
}, isLastFew:${lastFewChunks}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate completion statistics
|
||||
*/
|
||||
calculateCompletionStats(
|
||||
chunks: (ArrayBuffer | null)[],
|
||||
expectedChunksCount: number,
|
||||
expectedSize: number
|
||||
): {
|
||||
sequencedCount: number;
|
||||
currentTotalSize: number;
|
||||
isSequencedComplete: boolean;
|
||||
sizeComplete: boolean;
|
||||
isDataComplete: boolean;
|
||||
} {
|
||||
// Calculate current actual total received size
|
||||
const currentTotalSize = chunks.reduce((sum, chunk) => {
|
||||
return sum + (chunk instanceof ArrayBuffer ? chunk.byteLength : 0);
|
||||
}, 0);
|
||||
|
||||
// Count sequentially received chunks
|
||||
let sequencedCount = 0;
|
||||
for (let i = 0; i < expectedChunksCount; i++) {
|
||||
if (chunks[i] instanceof ArrayBuffer) {
|
||||
sequencedCount++;
|
||||
}
|
||||
}
|
||||
|
||||
const isSequencedComplete = sequencedCount === expectedChunksCount;
|
||||
const sizeComplete = currentTotalSize >= expectedSize;
|
||||
const isDataComplete = isSequencedComplete && sizeComplete;
|
||||
|
||||
return {
|
||||
sequencedCount,
|
||||
currentTotalSize,
|
||||
isSequencedComplete,
|
||||
sizeComplete,
|
||||
isDataComplete,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Log completion check details (for debugging)
|
||||
*/
|
||||
logCompletionCheck(
|
||||
fileName: string,
|
||||
stats: {
|
||||
sequencedCount: number;
|
||||
expectedChunksCount: number;
|
||||
currentTotalSize: number;
|
||||
expectedSize: number;
|
||||
isDataComplete: boolean;
|
||||
},
|
||||
chunks: (ArrayBuffer | null)[],
|
||||
initialOffset: number
|
||||
): void {
|
||||
if (!ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
return;
|
||||
}
|
||||
|
||||
const { sequencedCount, expectedChunksCount, currentTotalSize, expectedSize, isDataComplete } = stats;
|
||||
|
||||
// Only log at key moments to reduce noise
|
||||
if (
|
||||
isDataComplete ||
|
||||
sequencedCount % ReceptionConfig.DEBUG_CONFIG.PROGRESS_LOG_INTERVAL === 0 ||
|
||||
sequencedCount > expectedChunksCount - 10
|
||||
) {
|
||||
// Check last few chunks status
|
||||
const lastChunkIndex = expectedChunksCount - 1;
|
||||
const lastFewChunks = [];
|
||||
const startChunkIndex = ReceptionConfig.getChunkIndexFromOffset(initialOffset);
|
||||
|
||||
for (let i = Math.max(0, lastChunkIndex - 3); i <= lastChunkIndex; i++) {
|
||||
const chunk = chunks[i];
|
||||
const exists = chunk instanceof ArrayBuffer;
|
||||
const size = exists ? (chunk as ArrayBuffer).byteLength : 0;
|
||||
const absoluteIndex = startChunkIndex + i;
|
||||
lastFewChunks.push(`rel#${i}(abs#${absoluteIndex}):${exists}(${size})`);
|
||||
}
|
||||
|
||||
postLogToBackend(`[DEBUG-COMPLETE] Check completion - file:${fileName}`);
|
||||
postLogToBackend(
|
||||
`[DEBUG-COMPLETE] Chunks: received:${sequencedCount}/${expectedChunksCount}, isSequenceComplete:${stats.sequencedCount === expectedChunksCount}`
|
||||
);
|
||||
postLogToBackend(
|
||||
`[DEBUG-COMPLETE] Size: current:${currentTotalSize}, expected:${expectedSize}, sizeComplete:${currentTotalSize >= expectedSize}, diff:${
|
||||
expectedSize - currentTotalSize
|
||||
}`
|
||||
);
|
||||
postLogToBackend(
|
||||
`[DEBUG-COMPLETE] LastChunks: ${lastFewChunks.join(", ")}`
|
||||
);
|
||||
postLogToBackend(
|
||||
`[DEBUG-COMPLETE] IsDataComplete: ${isDataComplete}`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,280 @@
|
||||
import { CustomFile, fileMetadata } from "@/types/webrtc";
|
||||
import { ReceptionConfig } from "./ReceptionConfig";
|
||||
import { postLogToBackend } from "@/app/config/api";
|
||||
|
||||
const developmentEnv = process.env.NODE_ENV;
|
||||
|
||||
/**
|
||||
* 🚀 File assembly result interface
|
||||
*/
|
||||
export interface FileAssemblyResult {
|
||||
file: CustomFile;
|
||||
totalChunkSize: number;
|
||||
validChunks: number;
|
||||
storeUpdated: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 🚀 File assembler
|
||||
* Handles in-memory file assembly and validation
|
||||
*/
|
||||
export class FileAssembler {
|
||||
/**
|
||||
* Assemble file from chunks in memory
|
||||
*/
|
||||
async assembleFileFromChunks(
|
||||
chunks: (ArrayBuffer | null)[],
|
||||
meta: fileMetadata,
|
||||
currentFolderName: string | null,
|
||||
onFileReceived?: (file: CustomFile) => Promise<void>
|
||||
): Promise<FileAssemblyResult> {
|
||||
// Validate and count chunks
|
||||
let totalChunkSize = 0;
|
||||
let validChunks = 0;
|
||||
|
||||
chunks.forEach((chunk, index) => {
|
||||
if (chunk instanceof ArrayBuffer) {
|
||||
validChunks++;
|
||||
totalChunkSize += chunk.byteLength;
|
||||
}
|
||||
});
|
||||
|
||||
// Final verification
|
||||
const sizeDifference = meta.size - totalChunkSize;
|
||||
if (Math.abs(sizeDifference) > ReceptionConfig.VALIDATION_CONFIG.MAX_SIZE_DIFFERENCE_BYTES) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ❌ SIZE_MISMATCH - difference: ${sizeDifference} bytes (threshold: ${ReceptionConfig.VALIDATION_CONFIG.MAX_SIZE_DIFFERENCE_BYTES})`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Create file blob from valid chunks
|
||||
const validChunkBuffers = chunks.filter(
|
||||
(chunk) => chunk instanceof ArrayBuffer
|
||||
) as ArrayBuffer[];
|
||||
|
||||
const fileBlob = new Blob(validChunkBuffers, {
|
||||
type: meta.fileType,
|
||||
});
|
||||
|
||||
// Create File object
|
||||
const file = new File([fileBlob], meta.name, {
|
||||
type: meta.fileType,
|
||||
});
|
||||
|
||||
// Create CustomFile with additional properties
|
||||
const customFile = Object.assign(file, {
|
||||
fullName: meta.fullName,
|
||||
folderName: currentFolderName,
|
||||
}) as CustomFile;
|
||||
|
||||
// Store the file if callback is provided
|
||||
let storeUpdated = false;
|
||||
if (onFileReceived) {
|
||||
await onFileReceived(customFile);
|
||||
await Promise.resolve();
|
||||
await new Promise<void>((resolve) => setTimeout(() => resolve(), 0));
|
||||
storeUpdated = true;
|
||||
}
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ✅ File assembled - ${meta.name}, chunks: ${validChunks}/${chunks.length}, size: ${totalChunkSize}/${meta.size}, stored: ${storeUpdated}`
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
file: customFile,
|
||||
totalChunkSize,
|
||||
validChunks,
|
||||
storeUpdated,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate file assembly completeness
|
||||
*/
|
||||
validateAssembly(
|
||||
chunks: (ArrayBuffer | null)[],
|
||||
expectedSize: number,
|
||||
expectedChunksCount: number
|
||||
): {
|
||||
isComplete: boolean;
|
||||
validChunks: number;
|
||||
totalSize: number;
|
||||
missingChunks: number[];
|
||||
sizeDifference: number;
|
||||
} {
|
||||
let totalSize = 0;
|
||||
let validChunks = 0;
|
||||
const missingChunks: number[] = [];
|
||||
|
||||
chunks.forEach((chunk, index) => {
|
||||
if (chunk instanceof ArrayBuffer) {
|
||||
validChunks++;
|
||||
totalSize += chunk.byteLength;
|
||||
} else {
|
||||
missingChunks.push(index);
|
||||
}
|
||||
});
|
||||
|
||||
const sizeDifference = expectedSize - totalSize;
|
||||
const isComplete =
|
||||
validChunks === expectedChunksCount &&
|
||||
Math.abs(sizeDifference) <= ReceptionConfig.VALIDATION_CONFIG.MAX_SIZE_DIFFERENCE_BYTES;
|
||||
|
||||
return {
|
||||
isComplete,
|
||||
validChunks,
|
||||
totalSize,
|
||||
missingChunks,
|
||||
sizeDifference,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get assembly statistics for debugging
|
||||
*/
|
||||
getAssemblyStats(chunks: (ArrayBuffer | null)[]): {
|
||||
totalChunks: number;
|
||||
validChunks: number;
|
||||
nullChunks: number;
|
||||
totalSize: number;
|
||||
averageChunkSize: number;
|
||||
firstNullIndex: number | null;
|
||||
lastValidIndex: number | null;
|
||||
} {
|
||||
let validChunks = 0;
|
||||
let totalSize = 0;
|
||||
let firstNullIndex: number | null = null;
|
||||
let lastValidIndex: number | null = null;
|
||||
|
||||
chunks.forEach((chunk, index) => {
|
||||
if (chunk instanceof ArrayBuffer) {
|
||||
validChunks++;
|
||||
totalSize += chunk.byteLength;
|
||||
lastValidIndex = index;
|
||||
} else {
|
||||
if (firstNullIndex === null) {
|
||||
firstNullIndex = index;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
const averageChunkSize = validChunks > 0 ? totalSize / validChunks : 0;
|
||||
|
||||
return {
|
||||
totalChunks: chunks.length,
|
||||
validChunks,
|
||||
nullChunks: chunks.length - validChunks,
|
||||
totalSize,
|
||||
averageChunkSize,
|
||||
firstNullIndex,
|
||||
lastValidIndex,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create file download URL for in-memory files
|
||||
*/
|
||||
createDownloadUrl(file: File): string {
|
||||
return URL.createObjectURL(file);
|
||||
}
|
||||
|
||||
/**
|
||||
* Revoke file download URL to free memory
|
||||
*/
|
||||
revokeDownloadUrl(url: string): void {
|
||||
URL.revokeObjectURL(url);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get file type information
|
||||
*/
|
||||
getFileTypeInfo(file: File): {
|
||||
mimeType: string;
|
||||
extension: string;
|
||||
category: 'image' | 'video' | 'audio' | 'document' | 'archive' | 'other';
|
||||
} {
|
||||
const mimeType = file.type || 'application/octet-stream';
|
||||
const extension = file.name.split('.').pop()?.toLowerCase() || '';
|
||||
|
||||
let category: 'image' | 'video' | 'audio' | 'document' | 'archive' | 'other' = 'other';
|
||||
|
||||
if (mimeType.startsWith('image/')) {
|
||||
category = 'image';
|
||||
} else if (mimeType.startsWith('video/')) {
|
||||
category = 'video';
|
||||
} else if (mimeType.startsWith('audio/')) {
|
||||
category = 'audio';
|
||||
} else if (
|
||||
mimeType.includes('text/') ||
|
||||
mimeType.includes('application/pdf') ||
|
||||
mimeType.includes('application/msword') ||
|
||||
mimeType.includes('application/vnd.openxmlformats')
|
||||
) {
|
||||
category = 'document';
|
||||
} else if (
|
||||
mimeType.includes('zip') ||
|
||||
mimeType.includes('rar') ||
|
||||
mimeType.includes('tar') ||
|
||||
mimeType.includes('gzip')
|
||||
) {
|
||||
category = 'archive';
|
||||
}
|
||||
|
||||
return {
|
||||
mimeType,
|
||||
extension,
|
||||
category,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Estimate memory usage for file assembly
|
||||
*/
|
||||
estimateMemoryUsage(chunks: (ArrayBuffer | null)[]): {
|
||||
chunkMemoryUsage: number;
|
||||
estimatedBlobMemory: number;
|
||||
totalEstimatedMemory: number;
|
||||
} {
|
||||
const chunkMemoryUsage = chunks.reduce((sum, chunk) => {
|
||||
return sum + (chunk instanceof ArrayBuffer ? chunk.byteLength : 0);
|
||||
}, 0);
|
||||
|
||||
// Blob creation might temporarily double memory usage
|
||||
const estimatedBlobMemory = chunkMemoryUsage;
|
||||
const totalEstimatedMemory = chunkMemoryUsage + estimatedBlobMemory;
|
||||
|
||||
return {
|
||||
chunkMemoryUsage,
|
||||
estimatedBlobMemory,
|
||||
totalEstimatedMemory,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if file should be assembled in memory or streamed to disk
|
||||
*/
|
||||
shouldAssembleInMemory(
|
||||
fileSize: number,
|
||||
hasSaveDirectory: boolean,
|
||||
availableMemory?: number
|
||||
): boolean {
|
||||
// If we have a save directory and file is large, prefer disk
|
||||
if (hasSaveDirectory && fileSize >= ReceptionConfig.FILE_CONFIG.LARGE_FILE_THRESHOLD) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// If available memory is provided, check if we have enough
|
||||
if (availableMemory !== undefined) {
|
||||
// Need roughly 2x file size for assembly process
|
||||
const requiredMemory = fileSize * 2;
|
||||
return availableMemory > requiredMemory;
|
||||
}
|
||||
|
||||
// Default: assemble in memory for smaller files
|
||||
return fileSize < ReceptionConfig.FILE_CONFIG.LARGE_FILE_THRESHOLD;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,666 @@
|
||||
import WebRTC_Recipient from "../webrtc_Recipient";
|
||||
import { CustomFile, fileMetadata } from "@/types/webrtc";
|
||||
import { ReceptionStateManager } from "./ReceptionStateManager";
|
||||
import { MessageProcessor, MessageProcessorDelegate } from "./MessageProcessor";
|
||||
import { ChunkProcessor, ChunkProcessingResult } from "./ChunkProcessor";
|
||||
import { StreamingFileWriter, SequencedDiskWriter } from "./StreamingFileWriter";
|
||||
import { FileAssembler } from "./FileAssembler";
|
||||
import { ProgressReporter, ProgressCallback } from "./ProgressReporter";
|
||||
import { ReceptionConfig } from "./ReceptionConfig";
|
||||
import { postLogToBackend } from "@/app/config/api";
|
||||
|
||||
const developmentEnv = process.env.NODE_ENV;
|
||||
|
||||
/**
|
||||
* 🚀 File receive orchestrator
|
||||
* Main coordinator that integrates all reception modules
|
||||
*/
|
||||
export class FileReceiveOrchestrator implements MessageProcessorDelegate {
|
||||
private stateManager: ReceptionStateManager;
|
||||
private messageProcessor: MessageProcessor;
|
||||
private chunkProcessor: ChunkProcessor;
|
||||
private streamingFileWriter: StreamingFileWriter;
|
||||
private fileAssembler: FileAssembler;
|
||||
private progressReporter: ProgressReporter;
|
||||
|
||||
// Callbacks
|
||||
public onFileMetaReceived: ((meta: fileMetadata) => void) | undefined = undefined;
|
||||
public onStringReceived: ((str: string) => void) | undefined = undefined;
|
||||
public onFileReceived: ((file: CustomFile) => Promise<void>) | undefined = undefined;
|
||||
|
||||
constructor(private webrtcConnection: WebRTC_Recipient) {
|
||||
// Initialize all components
|
||||
this.stateManager = new ReceptionStateManager();
|
||||
this.chunkProcessor = new ChunkProcessor();
|
||||
this.streamingFileWriter = new StreamingFileWriter();
|
||||
this.fileAssembler = new FileAssembler();
|
||||
this.progressReporter = new ProgressReporter(this.stateManager);
|
||||
this.messageProcessor = new MessageProcessor(
|
||||
this.stateManager,
|
||||
webrtcConnection,
|
||||
{
|
||||
onFileMetaReceived: (meta: fileMetadata) => {
|
||||
if (this.onFileMetaReceived) {
|
||||
this.onFileMetaReceived(meta);
|
||||
}
|
||||
},
|
||||
onStringReceived: (str: string) => {
|
||||
if (this.onStringReceived) {
|
||||
this.onStringReceived(str);
|
||||
}
|
||||
},
|
||||
log: this.log.bind(this)
|
||||
}
|
||||
);
|
||||
|
||||
// Set up data handler
|
||||
this.setupDataHandler();
|
||||
|
||||
this.log("log", "FileReceiveOrchestrator initialized");
|
||||
}
|
||||
|
||||
// ===== Public API =====
|
||||
|
||||
/**
|
||||
* Set progress callback
|
||||
*/
|
||||
public setProgressCallback(callback: ProgressCallback): void {
|
||||
this.progressReporter.setProgressCallback(callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set save directory
|
||||
*/
|
||||
public setSaveDirectory(directory: FileSystemDirectoryHandle): Promise<void> {
|
||||
this.stateManager.setSaveDirectory(directory);
|
||||
this.streamingFileWriter.setSaveDirectory(directory);
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
/**
|
||||
* Request a single file from the peer
|
||||
*/
|
||||
public async requestFile(fileId: string, singleFile = true): Promise<void> {
|
||||
const activeReception = this.stateManager.getActiveFileReception();
|
||||
if (activeReception) {
|
||||
this.log("warn", "Another file reception is already in progress.");
|
||||
return;
|
||||
}
|
||||
|
||||
if (singleFile) {
|
||||
this.stateManager.setCurrentFolderName(null);
|
||||
}
|
||||
|
||||
const fileInfo = this.stateManager.getFileMetadata(fileId);
|
||||
if (!fileInfo) {
|
||||
this.fireError("File info not found for the requested fileId", { fileId });
|
||||
return;
|
||||
}
|
||||
|
||||
const shouldSaveToDisk = ReceptionConfig.shouldSaveToDisk(
|
||||
fileInfo.size,
|
||||
this.streamingFileWriter.hasSaveDirectory()
|
||||
);
|
||||
|
||||
// Set save type at the beginning to prevent race conditions
|
||||
this.stateManager.setSaveType(fileInfo.fileId, shouldSaveToDisk);
|
||||
const currentFolderName = this.stateManager.getCurrentFolderName();
|
||||
if (currentFolderName) {
|
||||
this.stateManager.setSaveType(currentFolderName, shouldSaveToDisk);
|
||||
}
|
||||
|
||||
let offset = 0;
|
||||
if (shouldSaveToDisk && this.streamingFileWriter.hasSaveDirectory()) {
|
||||
try {
|
||||
offset = await this.streamingFileWriter.getPartialFileSize(
|
||||
fileInfo.name,
|
||||
fileInfo.fullName
|
||||
);
|
||||
|
||||
if (offset === fileInfo.size) {
|
||||
this.log("log", "File already fully downloaded.", { fileId });
|
||||
this.progressReporter.reportFileComplete(fileId);
|
||||
return;
|
||||
}
|
||||
this.log("log", `Resuming file from offset: ${offset}`, { fileId });
|
||||
} catch (e) {
|
||||
this.log("log", "Partial file not found, starting from scratch.", { fileId });
|
||||
offset = 0;
|
||||
}
|
||||
}
|
||||
|
||||
const expectedChunksCount = ReceptionConfig.calculateExpectedChunks(
|
||||
fileInfo.size,
|
||||
offset
|
||||
);
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
const totalChunks = ReceptionConfig.calculateTotalChunks(fileInfo.size);
|
||||
const startChunkIndex = ReceptionConfig.getChunkIndexFromOffset(offset);
|
||||
|
||||
postLogToBackend(`[DEBUG-CHUNKS] File: ${fileInfo.name}`);
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] File size: ${fileInfo.size}, offset: ${offset}`
|
||||
);
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] Total chunks in file: ${totalChunks} (0-${totalChunks - 1})`
|
||||
);
|
||||
postLogToBackend(`[DEBUG-CHUNKS] Start chunk index: ${startChunkIndex}`);
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] Expected chunks: ${expectedChunksCount}`
|
||||
);
|
||||
}
|
||||
|
||||
const receptionPromise = this.stateManager.startFileReception(
|
||||
fileInfo,
|
||||
expectedChunksCount,
|
||||
offset
|
||||
);
|
||||
|
||||
if (shouldSaveToDisk) {
|
||||
await this.createDiskWriteStream(fileInfo, offset);
|
||||
}
|
||||
|
||||
// Send file request
|
||||
const success = this.messageProcessor.sendFileRequest(fileId, offset);
|
||||
if (!success) {
|
||||
this.stateManager.failFileReception(new Error("Failed to send file request"));
|
||||
return;
|
||||
}
|
||||
|
||||
return receptionPromise;
|
||||
}
|
||||
|
||||
/**
|
||||
* Request all files belonging to a folder from the peer
|
||||
*/
|
||||
public async requestFolder(folderName: string): Promise<void> {
|
||||
const folderProgress = this.stateManager.getFolderProgress(folderName);
|
||||
if (!folderProgress || folderProgress.fileIds.length === 0) {
|
||||
this.log("warn", "No files found for the requested folder.", { folderName });
|
||||
return;
|
||||
}
|
||||
|
||||
// Pre-calculate total size of already downloaded parts
|
||||
let initialFolderReceivedSize = 0;
|
||||
if (this.streamingFileWriter.hasSaveDirectory()) {
|
||||
for (const fileId of folderProgress.fileIds) {
|
||||
const fileInfo = this.stateManager.getFileMetadata(fileId);
|
||||
if (fileInfo) {
|
||||
try {
|
||||
const partialSize = await this.streamingFileWriter.getPartialFileSize(
|
||||
fileInfo.name,
|
||||
fileInfo.fullName
|
||||
);
|
||||
initialFolderReceivedSize += partialSize;
|
||||
} catch (e) {
|
||||
// File doesn't exist, so its size is 0
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.stateManager.setFolderReceivedSize(folderName, initialFolderReceivedSize);
|
||||
this.log(
|
||||
"log",
|
||||
`Requesting folder, initial received size: ${initialFolderReceivedSize}`,
|
||||
{ folderName }
|
||||
);
|
||||
|
||||
this.stateManager.setCurrentFolderName(folderName);
|
||||
|
||||
for (const fileId of folderProgress.fileIds) {
|
||||
try {
|
||||
await this.requestFile(fileId, false);
|
||||
} catch (error) {
|
||||
this.fireError(
|
||||
`Failed to receive file ${fileId} in folder ${folderName}`,
|
||||
{ error }
|
||||
);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
this.stateManager.setCurrentFolderName(null);
|
||||
|
||||
// Send folder completion message
|
||||
const completedFileIds = folderProgress.fileIds.filter(() => true); // Assume all succeeded
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 📁 All files in folder completed - ${folderName}, files: ${completedFileIds.length}/${folderProgress.fileIds.length}`
|
||||
);
|
||||
}
|
||||
|
||||
this.messageProcessor.sendFolderReceiveComplete(folderName, completedFileIds, true);
|
||||
}
|
||||
|
||||
// ===== MessageProcessorDelegate Implementation =====
|
||||
|
||||
// Note: These are implemented as properties, not methods, to avoid infinite recursion
|
||||
|
||||
public log(
|
||||
level: "log" | "warn" | "error",
|
||||
message: string,
|
||||
context?: Record<string, any>
|
||||
): void {
|
||||
const prefix = `[FileReceiveOrchestrator]`;
|
||||
console[level](prefix, message, context || "");
|
||||
}
|
||||
|
||||
// ===== Internal Methods =====
|
||||
|
||||
/**
|
||||
* Set up data handler
|
||||
*/
|
||||
private setupDataHandler(): void {
|
||||
this.webrtcConnection.onDataReceived = async (data, peerId) => {
|
||||
const binaryData = await this.messageProcessor.handleReceivedMessage(data, peerId);
|
||||
|
||||
if (binaryData) {
|
||||
// Handle binary chunk data
|
||||
await this.handleBinaryChunkData(binaryData);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle binary chunk data
|
||||
*/
|
||||
private async handleBinaryChunkData(data: any): Promise<void> {
|
||||
const activeReception = this.stateManager.getActiveFileReception();
|
||||
if (!activeReception) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ERROR: Received file chunk but no active file reception!`
|
||||
);
|
||||
}
|
||||
this.fireError("Received a file chunk without an active file reception.");
|
||||
return;
|
||||
}
|
||||
|
||||
// Convert to ArrayBuffer
|
||||
const arrayBuffer = await this.chunkProcessor.convertToArrayBuffer(data);
|
||||
if (!arrayBuffer) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ERROR: Failed to convert binary data to ArrayBuffer`
|
||||
);
|
||||
}
|
||||
this.fireError("Received unsupported binary data format", {
|
||||
dataType: Object.prototype.toString.call(data),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
await this.handleEmbeddedChunkPacket(arrayBuffer);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle embedded chunk packet
|
||||
*/
|
||||
private async handleEmbeddedChunkPacket(arrayBuffer: ArrayBuffer): Promise<void> {
|
||||
const parsed = this.chunkProcessor.parseEmbeddedChunkPacket(arrayBuffer);
|
||||
if (!parsed) {
|
||||
this.fireError("Failed to parse embedded chunk packet");
|
||||
return;
|
||||
}
|
||||
|
||||
const { chunkMeta, chunkData } = parsed;
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (!reception) {
|
||||
console.log(
|
||||
`[FileReceiveOrchestrator] Ignoring chunk ${chunkMeta.chunkIndex} - file reception already closed`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Validate chunk
|
||||
const validation = this.chunkProcessor.validateChunk(
|
||||
chunkMeta,
|
||||
reception.meta.fileId,
|
||||
reception.expectedChunksCount,
|
||||
reception.initialOffset
|
||||
);
|
||||
|
||||
if (!validation.isValid) {
|
||||
this.log("warn", "Chunk validation failed", {
|
||||
errors: validation.errors,
|
||||
chunkIndex: chunkMeta.chunkIndex,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
// Process chunk indices
|
||||
const result = this.chunkProcessor.processReceivedChunk(
|
||||
chunkMeta,
|
||||
chunkData,
|
||||
reception.initialOffset
|
||||
);
|
||||
|
||||
if (!result) {
|
||||
this.fireError("Failed to process received chunk");
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if chunk index is valid
|
||||
if (!this.chunkProcessor.isChunkIndexValid(
|
||||
result.relativeChunkIndex,
|
||||
reception.expectedChunksCount
|
||||
)) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-CHUNKS] ❌ Invalid relative chunk index - absolute:${result.absoluteChunkIndex}, relative:${result.relativeChunkIndex}, arraySize:${reception.chunks.length}`
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Store chunk
|
||||
reception.chunks[result.relativeChunkIndex] = result.chunkData;
|
||||
reception.chunkSequenceMap.set(result.absoluteChunkIndex, true);
|
||||
reception.receivedChunksCount++;
|
||||
|
||||
// Update progress
|
||||
this.progressReporter.updateFileProgress(
|
||||
result.chunkData.byteLength,
|
||||
reception.meta.fileId,
|
||||
reception.meta.size
|
||||
);
|
||||
|
||||
// Handle disk writing if needed
|
||||
if (reception.sequencedWriter) {
|
||||
this.chunkProcessor.logChunkDetails(
|
||||
result,
|
||||
reception.expectedChunksCount,
|
||||
reception.sequencedWriter.expectedIndex
|
||||
);
|
||||
|
||||
await reception.sequencedWriter.writeChunk(
|
||||
result.absoluteChunkIndex,
|
||||
result.chunkData
|
||||
);
|
||||
}
|
||||
|
||||
await this.checkAndAutoFinalize();
|
||||
}
|
||||
|
||||
/**
|
||||
* Check and auto-finalize file reception
|
||||
*/
|
||||
private async checkAndAutoFinalize(): Promise<void> {
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (!reception || reception.isFinalized) return;
|
||||
|
||||
const expectedSize = reception.meta.size - reception.initialOffset;
|
||||
const stats = this.chunkProcessor.calculateCompletionStats(
|
||||
reception.chunks,
|
||||
reception.expectedChunksCount,
|
||||
expectedSize
|
||||
);
|
||||
|
||||
// Log completion check details
|
||||
this.chunkProcessor.logCompletionCheck(
|
||||
reception.meta.name,
|
||||
{
|
||||
sequencedCount: stats.sequencedCount,
|
||||
expectedChunksCount: reception.expectedChunksCount,
|
||||
currentTotalSize: stats.currentTotalSize,
|
||||
expectedSize,
|
||||
isDataComplete: stats.isDataComplete,
|
||||
},
|
||||
reception.chunks,
|
||||
reception.initialOffset
|
||||
);
|
||||
|
||||
if (stats.isDataComplete) {
|
||||
reception.isFinalized = true;
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-COMPLETE] ✅ Starting finalization - isDataComplete:${stats.isDataComplete}`
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
await this.finalizeFileReceive();
|
||||
this.stateManager.completeFileReception();
|
||||
} catch (error) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(`[DEBUG] ❌ Auto-finalize ERROR: ${error}`);
|
||||
}
|
||||
this.stateManager.failFileReception(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalize file reception
|
||||
*/
|
||||
private async finalizeFileReceive(): Promise<void> {
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (!reception) return;
|
||||
|
||||
if (reception.writeStream) {
|
||||
await this.finalizeLargeFileReceive();
|
||||
} else {
|
||||
await this.finalizeMemoryFileReceive();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalize large file reception (disk-based)
|
||||
*/
|
||||
private async finalizeLargeFileReceive(): Promise<void> {
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (!reception?.writeStream || !reception.fileHandle) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] ❌ Cannot finalize - missing writeStream:${!!reception?.writeStream} or fileHandle:${!!reception?.fileHandle}`
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] 🚀 Starting finalization for ${reception.meta.name}`
|
||||
);
|
||||
}
|
||||
|
||||
// Finalize using StreamingFileWriter
|
||||
if (reception.sequencedWriter && reception.writeStream) {
|
||||
await this.streamingFileWriter.finalizeWrite(
|
||||
reception.sequencedWriter,
|
||||
reception.writeStream,
|
||||
reception.meta.name
|
||||
);
|
||||
}
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] ✅ LARGE_FILE finalized successfully - ${reception.meta.name}`
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] ❌ Error during finalization: ${error}`
|
||||
);
|
||||
}
|
||||
this.fireError("Error finalizing large file", { error });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalize memory file reception
|
||||
*/
|
||||
private async finalizeMemoryFileReceive(): Promise<void> {
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (!reception) return;
|
||||
|
||||
const currentFolderName = this.stateManager.getCurrentFolderName();
|
||||
const result = await this.fileAssembler.assembleFileFromChunks(
|
||||
reception.chunks,
|
||||
reception.meta,
|
||||
currentFolderName,
|
||||
this.onFileReceived
|
||||
);
|
||||
|
||||
// Send completion confirmation
|
||||
this.messageProcessor.sendFileReceiveComplete(
|
||||
reception.meta.fileId,
|
||||
result.totalChunkSize,
|
||||
result.validChunks,
|
||||
result.storeUpdated
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create disk write stream
|
||||
*/
|
||||
private async createDiskWriteStream(
|
||||
meta: fileMetadata,
|
||||
offset: number
|
||||
): Promise<void> {
|
||||
try {
|
||||
const { fileHandle, writeStream, sequencedWriter } =
|
||||
await this.streamingFileWriter.createWriteStream(
|
||||
meta.name,
|
||||
meta.fullName,
|
||||
offset
|
||||
);
|
||||
|
||||
this.stateManager.updateActiveFileReception({
|
||||
fileHandle,
|
||||
writeStream,
|
||||
sequencedWriter,
|
||||
});
|
||||
} catch (err) {
|
||||
this.fireError("Failed to create file on disk", {
|
||||
err,
|
||||
fileName: meta.name,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Error handling
|
||||
*/
|
||||
private fireError(message: string, context?: Record<string, any>) {
|
||||
if (this.webrtcConnection.fireError) {
|
||||
// @ts-ignore
|
||||
this.webrtcConnection.fireError(message, {
|
||||
...context,
|
||||
component: "FileReceiveOrchestrator",
|
||||
});
|
||||
} else {
|
||||
this.log("error", message, context);
|
||||
}
|
||||
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (reception) {
|
||||
// Clean up resources on error
|
||||
if (reception.sequencedWriter) {
|
||||
reception.sequencedWriter.close().catch((err: any) => {
|
||||
this.log(
|
||||
"error",
|
||||
"Error closing sequenced writer during error cleanup",
|
||||
{ err }
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
this.stateManager.failFileReception(new Error(message));
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Lifecycle Management =====
|
||||
|
||||
/**
|
||||
* Graceful shutdown
|
||||
*/
|
||||
public gracefulShutdown(reason: string = "CONNECTION_LOST"): void {
|
||||
this.log("log", `Graceful shutdown initiated: ${reason}`);
|
||||
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (reception?.sequencedWriter && reception?.writeStream) {
|
||||
this.log(
|
||||
"log",
|
||||
"Attempting to gracefully close streams on shutdown."
|
||||
);
|
||||
|
||||
// Close sequenced writer and write stream
|
||||
reception.sequencedWriter.close().catch((err: any) => {
|
||||
this.log("error", "Error closing sequenced writer during graceful shutdown", { err });
|
||||
});
|
||||
|
||||
reception.writeStream.close().catch((err: any) => {
|
||||
this.log("error", "Error closing stream during graceful shutdown", { err });
|
||||
});
|
||||
}
|
||||
|
||||
this.stateManager.gracefulCleanup();
|
||||
this.log("log", "Graceful shutdown completed");
|
||||
}
|
||||
|
||||
/**
|
||||
* Force reset all internal states
|
||||
*/
|
||||
public forceReset(): void {
|
||||
this.log("log", "Force resetting FileReceiveOrchestrator state");
|
||||
|
||||
const reception = this.stateManager.getActiveFileReception();
|
||||
if (reception?.sequencedWriter && reception?.writeStream) {
|
||||
reception.sequencedWriter.close().catch(console.error);
|
||||
reception.writeStream.close().catch(console.error);
|
||||
}
|
||||
|
||||
this.stateManager.forceReset();
|
||||
this.progressReporter.resetAllProgress();
|
||||
this.log("log", "FileReceiveOrchestrator state force reset completed");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get transfer statistics
|
||||
*/
|
||||
public getTransferStats() {
|
||||
return {
|
||||
stateManager: this.stateManager.getStateStats(),
|
||||
progressReporter: this.progressReporter.getProgressStats(),
|
||||
messageProcessor: this.messageProcessor.getMessageStats(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get save type information (for backward compatibility)
|
||||
*/
|
||||
public getSaveType(): Record<string, boolean> {
|
||||
return this.stateManager.saveType;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get pending files metadata (for backward compatibility)
|
||||
*/
|
||||
public getPendingFilesMeta(): Map<string, fileMetadata> {
|
||||
return this.stateManager.getAllFileMetadata();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get folder progresses (for backward compatibility)
|
||||
*/
|
||||
public getFolderProgresses(): Record<string, any> {
|
||||
return this.stateManager.getAllFolderProgresses();
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up all resources
|
||||
*/
|
||||
public cleanup(): void {
|
||||
this.stateManager.gracefulCleanup();
|
||||
this.progressReporter.cleanup();
|
||||
this.messageProcessor.cleanup();
|
||||
this.log("log", "FileReceiveOrchestrator cleaned up");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,302 @@
|
||||
import {
|
||||
WebRTCMessage,
|
||||
fileMetadata,
|
||||
StringMetadata,
|
||||
StringChunk,
|
||||
FileRequest,
|
||||
FileReceiveComplete,
|
||||
FolderReceiveComplete,
|
||||
FileHandlers,
|
||||
} from "@/types/webrtc";
|
||||
import { ReceptionStateManager } from "./ReceptionStateManager";
|
||||
import { ReceptionConfig } from "./ReceptionConfig";
|
||||
import { postLogToBackend } from "@/app/config/api";
|
||||
import WebRTC_Recipient from "../webrtc_Recipient";
|
||||
|
||||
const developmentEnv = process.env.NODE_ENV;
|
||||
|
||||
/**
|
||||
* 🚀 Message processor delegate interface
|
||||
*/
|
||||
export interface MessageProcessorDelegate {
|
||||
onFileMetaReceived?: (meta: fileMetadata) => void;
|
||||
onStringReceived?: (str: string) => void;
|
||||
log(level: "log" | "warn" | "error", message: string, context?: Record<string, any>): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* 🚀 Message processor
|
||||
* Handles WebRTC message routing, processing, and communication
|
||||
*/
|
||||
export class MessageProcessor {
|
||||
private fileHandlers: FileHandlers;
|
||||
|
||||
constructor(
|
||||
private stateManager: ReceptionStateManager,
|
||||
private webrtcConnection: WebRTC_Recipient,
|
||||
private delegate: MessageProcessorDelegate
|
||||
) {
|
||||
this.fileHandlers = {
|
||||
string: this.handleReceivedStringChunk.bind(this),
|
||||
stringMetadata: this.handleStringMetadata.bind(this),
|
||||
fileMeta: this.handleFileMetadata.bind(this),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle received WebRTC message
|
||||
*/
|
||||
async handleReceivedMessage(
|
||||
data: string | ArrayBuffer | any,
|
||||
peerId: string
|
||||
): Promise<ArrayBuffer | null> {
|
||||
this.stateManager.setCurrentPeerId(peerId);
|
||||
|
||||
if (typeof data === "string") {
|
||||
try {
|
||||
const parsedData = JSON.parse(data) as WebRTCMessage;
|
||||
const handler = this.fileHandlers[parsedData.type as keyof FileHandlers];
|
||||
|
||||
if (handler) {
|
||||
await handler(parsedData as any, peerId);
|
||||
} else {
|
||||
this.delegate.log(
|
||||
"warn",
|
||||
`Handler not found for message type: ${parsedData.type}`,
|
||||
{ peerId }
|
||||
);
|
||||
}
|
||||
return null; // String messages don't return binary data
|
||||
} catch (error) {
|
||||
this.delegate.log("error", "Error parsing received JSON data", { error, peerId });
|
||||
return null;
|
||||
}
|
||||
} else {
|
||||
// Return binary data for chunk processing
|
||||
return data;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle file metadata message
|
||||
*/
|
||||
private handleFileMetadata(metadata: fileMetadata): void {
|
||||
const isNewMetadata = this.stateManager.addFileMetadata(metadata);
|
||||
|
||||
if (!isNewMetadata) {
|
||||
return; // Ignore if already received
|
||||
}
|
||||
|
||||
if (this.delegate.onFileMetaReceived) {
|
||||
this.delegate.onFileMetaReceived(metadata);
|
||||
} else {
|
||||
this.delegate.log(
|
||||
"error",
|
||||
"onFileMetaReceived callback not set",
|
||||
{ fileId: metadata.fileId }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle string metadata message
|
||||
*/
|
||||
private handleStringMetadata(metadata: StringMetadata): void {
|
||||
this.stateManager.startStringReception(metadata.length);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle received string chunk message
|
||||
*/
|
||||
private handleReceivedStringChunk(data: StringChunk): void {
|
||||
const activeStringReception = this.stateManager.getActiveStringReception();
|
||||
if (!activeStringReception) {
|
||||
this.delegate.log("warn", "Received string chunk without active reception");
|
||||
return;
|
||||
}
|
||||
|
||||
this.stateManager.updateStringReceptionChunk(data.index, data.chunk);
|
||||
|
||||
// Check if string reception is complete
|
||||
if (activeStringReception.receivedChunks === data.total) {
|
||||
const fullString = this.stateManager.completeStringReception();
|
||||
if (fullString && this.delegate.onStringReceived) {
|
||||
this.delegate.onStringReceived(fullString);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Send file request message
|
||||
*/
|
||||
sendFileRequest(fileId: string, offset: number = 0): boolean {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
if (!peerId) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ERROR: Cannot send fileRequest - no peerId available!`
|
||||
);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
const request: FileRequest = { type: "fileRequest", fileId, offset };
|
||||
const success = this.webrtcConnection.sendData(JSON.stringify(request), peerId);
|
||||
|
||||
if (success) {
|
||||
this.delegate.log("log", "Sent fileRequest", { request, peerId });
|
||||
} else {
|
||||
this.delegate.log("error", "Failed to send fileRequest", { request, peerId });
|
||||
}
|
||||
|
||||
return success;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send file receive complete message
|
||||
*/
|
||||
sendFileReceiveComplete(
|
||||
fileId: string,
|
||||
receivedSize: number,
|
||||
receivedChunks: number,
|
||||
storeUpdated: boolean
|
||||
): boolean {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
if (!peerId) {
|
||||
this.delegate.log("warn", "Cannot send file receive complete - no peer ID");
|
||||
return false;
|
||||
}
|
||||
|
||||
const completeMessage: FileReceiveComplete = {
|
||||
type: "fileReceiveComplete",
|
||||
fileId,
|
||||
receivedSize,
|
||||
receivedChunks,
|
||||
storeUpdated,
|
||||
};
|
||||
|
||||
const success = this.webrtcConnection.sendData(
|
||||
JSON.stringify(completeMessage),
|
||||
peerId
|
||||
);
|
||||
|
||||
if (success) {
|
||||
this.delegate.log("log", "Sent file receive complete", {
|
||||
fileId,
|
||||
receivedSize,
|
||||
receivedChunks,
|
||||
storeUpdated,
|
||||
});
|
||||
} else {
|
||||
this.delegate.log("error", "Failed to send file receive complete", {
|
||||
fileId,
|
||||
peerId,
|
||||
});
|
||||
}
|
||||
|
||||
return success;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send folder receive complete message
|
||||
*/
|
||||
sendFolderReceiveComplete(
|
||||
folderName: string,
|
||||
completedFileIds: string[],
|
||||
allStoreUpdated: boolean
|
||||
): boolean {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
if (!peerId) {
|
||||
this.delegate.log("warn", "Cannot send folder receive complete - no peer ID");
|
||||
return false;
|
||||
}
|
||||
|
||||
const completeMessage: FolderReceiveComplete = {
|
||||
type: "folderReceiveComplete",
|
||||
folderName,
|
||||
completedFileIds,
|
||||
allStoreUpdated,
|
||||
};
|
||||
|
||||
const success = this.webrtcConnection.sendData(
|
||||
JSON.stringify(completeMessage),
|
||||
peerId
|
||||
);
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 📤 Sent folderReceiveComplete - folderName: ${folderName}, completedFiles: ${completedFileIds.length}, allStoreUpdated: ${allStoreUpdated}, success: ${success}`
|
||||
);
|
||||
}
|
||||
|
||||
if (success) {
|
||||
this.delegate.log("log", "Sent folder receive complete", {
|
||||
folderName,
|
||||
completedFiles: completedFileIds.length,
|
||||
allStoreUpdated,
|
||||
});
|
||||
} else {
|
||||
this.delegate.log("error", "Failed to send folder receive complete", {
|
||||
folderName,
|
||||
peerId,
|
||||
});
|
||||
}
|
||||
|
||||
return success;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add Firefox compatibility delay
|
||||
*/
|
||||
async addFirefoxDelay(): Promise<void> {
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, ReceptionConfig.NETWORK_CONFIG.FIREFOX_COMPATIBILITY_DELAY)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get message processing statistics
|
||||
*/
|
||||
getMessageStats(): {
|
||||
handledMessages: number;
|
||||
lastMessageTime: number | null;
|
||||
currentPeerId: string;
|
||||
} {
|
||||
return {
|
||||
handledMessages: 0, // TODO: Implement message counting if needed
|
||||
lastMessageTime: null, // TODO: Record last message time if needed
|
||||
currentPeerId: this.stateManager.getCurrentPeerId(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if connection is available
|
||||
*/
|
||||
isConnectionAvailable(): boolean {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
return !!peerId && !!this.webrtcConnection;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current peer connection info
|
||||
*/
|
||||
getPeerConnectionInfo(): {
|
||||
peerId: string;
|
||||
isConnected: boolean;
|
||||
} {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
return {
|
||||
peerId,
|
||||
isConnected: this.isConnectionAvailable(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up resources
|
||||
*/
|
||||
cleanup(): void {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend("[DEBUG] 🧹 MessageProcessor cleaned up");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,309 @@
|
||||
import { SpeedCalculator } from "@/lib/speedCalculator";
|
||||
import { ReceptionStateManager } from "./ReceptionStateManager";
|
||||
import { ReceptionConfig } from "./ReceptionConfig";
|
||||
import { postLogToBackend } from "@/app/config/api";
|
||||
|
||||
const developmentEnv = process.env.NODE_ENV;
|
||||
|
||||
/**
|
||||
* 🚀 Progress callback type
|
||||
*/
|
||||
export type ProgressCallback = (fileId: string, progress: number, speed: number) => void;
|
||||
|
||||
/**
|
||||
* 🚀 Progress statistics interface
|
||||
*/
|
||||
export interface ProgressStats {
|
||||
fileProgress: Record<string, number>;
|
||||
folderProgress: Record<string, number>;
|
||||
currentSpeed: number;
|
||||
averageSpeed: number;
|
||||
totalBytesReceived: number;
|
||||
estimatedTimeRemaining: number | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 🚀 Progress reporter
|
||||
* Handles progress calculation, speed tracking, and progress callback management
|
||||
*/
|
||||
export class ProgressReporter {
|
||||
private speedCalculator: SpeedCalculator;
|
||||
private progressCallback: ProgressCallback | null = null;
|
||||
|
||||
// Progress tracking
|
||||
private fileProgressMap = new Map<string, number>();
|
||||
private folderProgressMap = new Map<string, number>();
|
||||
private lastProgressUpdate = new Map<string, number>();
|
||||
|
||||
constructor(private stateManager: ReceptionStateManager) {
|
||||
this.speedCalculator = new SpeedCalculator();
|
||||
}
|
||||
|
||||
/**
|
||||
* Set progress callback
|
||||
*/
|
||||
setProgressCallback(callback: ProgressCallback): void {
|
||||
this.progressCallback = callback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Update file reception progress
|
||||
*/
|
||||
updateFileProgress(
|
||||
byteLength: number,
|
||||
fileId: string,
|
||||
fileSize: number
|
||||
): void {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
if (!peerId) return;
|
||||
|
||||
const activeReception = this.stateManager.getActiveFileReception();
|
||||
if (!activeReception) return;
|
||||
|
||||
// Update received size
|
||||
activeReception.receivedSize += byteLength;
|
||||
const totalReceived = activeReception.initialOffset + activeReception.receivedSize;
|
||||
|
||||
const currentFolderName = this.stateManager.getCurrentFolderName();
|
||||
|
||||
if (currentFolderName) {
|
||||
// Update folder progress
|
||||
this.updateFolderProgress(currentFolderName, byteLength, peerId);
|
||||
} else {
|
||||
// Update individual file progress
|
||||
this.speedCalculator.updateSendSpeed(peerId, totalReceived);
|
||||
const speed = this.speedCalculator.getSendSpeed(peerId);
|
||||
const progress = fileSize > 0 ? totalReceived / fileSize : 0;
|
||||
|
||||
// Store progress for statistics
|
||||
this.fileProgressMap.set(fileId, progress);
|
||||
|
||||
// Throttle progress callbacks to avoid overwhelming the UI
|
||||
const now = Date.now();
|
||||
const lastUpdate = this.lastProgressUpdate.get(fileId) || 0;
|
||||
const shouldUpdate = now - lastUpdate > 100; // Update at most every 100ms
|
||||
|
||||
if (shouldUpdate || progress >= 1) {
|
||||
this.progressCallback?.(fileId, progress, speed);
|
||||
this.lastProgressUpdate.set(fileId, now);
|
||||
}
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_PROGRESS_LOGGING && progress >= 1) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 📈 File progress 100% - ${fileId}, speed: ${(speed / 1024 / 1024).toFixed(1)}MB/s`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update folder reception progress
|
||||
*/
|
||||
private updateFolderProgress(
|
||||
folderName: string,
|
||||
byteLength: number,
|
||||
peerId: string
|
||||
): void {
|
||||
// Update folder received size in state manager
|
||||
this.stateManager.updateFolderReceivedSize(folderName, byteLength);
|
||||
|
||||
const folderProgress = this.stateManager.getFolderProgress(folderName);
|
||||
if (!folderProgress) return;
|
||||
|
||||
this.speedCalculator.updateSendSpeed(peerId, folderProgress.receivedSize);
|
||||
const speed = this.speedCalculator.getSendSpeed(peerId);
|
||||
const progress = folderProgress.totalSize > 0
|
||||
? folderProgress.receivedSize / folderProgress.totalSize
|
||||
: 0;
|
||||
|
||||
// Store progress for statistics
|
||||
this.folderProgressMap.set(folderName, progress);
|
||||
|
||||
// Throttle folder progress callbacks
|
||||
const now = Date.now();
|
||||
const lastUpdate = this.lastProgressUpdate.get(folderName) || 0;
|
||||
const shouldUpdate = now - lastUpdate > 200; // Update folders less frequently
|
||||
|
||||
if (shouldUpdate || progress >= 1) {
|
||||
this.progressCallback?.(folderName, progress, speed);
|
||||
this.lastProgressUpdate.set(folderName, now);
|
||||
}
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_PROGRESS_LOGGING && progress >= 1) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 📈 Folder progress 100% - ${folderName}, speed: ${(speed / 1024 / 1024).toFixed(1)}MB/s`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Report file completion (100% progress)
|
||||
*/
|
||||
reportFileComplete(fileId: string): void {
|
||||
if (!this.progressCallback) return;
|
||||
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
if (!peerId) return;
|
||||
|
||||
// Get final speed and report 100% progress
|
||||
const speed = this.speedCalculator.getSendSpeed(peerId);
|
||||
this.progressCallback(fileId, 1, speed);
|
||||
this.fileProgressMap.set(fileId, 1);
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_PROGRESS_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ✅ File completion reported - ${fileId}, final speed: ${(speed / 1024 / 1024).toFixed(1)}MB/s`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Report folder completion (100% progress)
|
||||
*/
|
||||
reportFolderComplete(folderName: string): void {
|
||||
if (!this.progressCallback) return;
|
||||
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
if (!peerId) return;
|
||||
|
||||
// Get final speed and report 100% progress
|
||||
const speed = this.speedCalculator.getSendSpeed(peerId);
|
||||
this.progressCallback(folderName, 1, speed);
|
||||
this.folderProgressMap.set(folderName, 1);
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_PROGRESS_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ✅ Folder completion reported - ${folderName}, final speed: ${(speed / 1024 / 1024).toFixed(1)}MB/s`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current progress for a file or folder
|
||||
*/
|
||||
getCurrentProgress(id: string): number {
|
||||
return this.fileProgressMap.get(id) || this.folderProgressMap.get(id) || 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current speed for peer
|
||||
*/
|
||||
getCurrentSpeed(): number {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
return peerId ? this.speedCalculator.getSendSpeed(peerId) : 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get detailed progress statistics
|
||||
*/
|
||||
getProgressStats(): ProgressStats {
|
||||
const peerId = this.stateManager.getCurrentPeerId();
|
||||
const currentSpeed = peerId ? this.speedCalculator.getSendSpeed(peerId) : 0;
|
||||
const averageSpeed = currentSpeed; // SpeedCalculator doesn't have getAverageSpeed method
|
||||
|
||||
// Calculate total bytes received
|
||||
let totalBytesReceived = 0;
|
||||
const activeReception = this.stateManager.getActiveFileReception();
|
||||
if (activeReception) {
|
||||
totalBytesReceived = activeReception.initialOffset + activeReception.receivedSize;
|
||||
}
|
||||
|
||||
// Estimate time remaining
|
||||
let estimatedTimeRemaining: number | null = null;
|
||||
if (activeReception && currentSpeed > 0) {
|
||||
const remainingBytes = activeReception.meta.size - totalBytesReceived;
|
||||
if (remainingBytes > 0) {
|
||||
estimatedTimeRemaining = remainingBytes / currentSpeed; // seconds
|
||||
}
|
||||
}
|
||||
|
||||
const fileProgress: Record<string, number> = {};
|
||||
this.fileProgressMap.forEach((progress, fileId) => {
|
||||
fileProgress[fileId] = progress;
|
||||
});
|
||||
|
||||
const folderProgress: Record<string, number> = {};
|
||||
this.folderProgressMap.forEach((progress, folderName) => {
|
||||
folderProgress[folderName] = progress;
|
||||
});
|
||||
|
||||
return {
|
||||
fileProgress,
|
||||
folderProgress,
|
||||
currentSpeed,
|
||||
averageSpeed,
|
||||
totalBytesReceived,
|
||||
estimatedTimeRemaining,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset progress for a specific file or folder
|
||||
*/
|
||||
resetProgress(id: string): void {
|
||||
this.fileProgressMap.delete(id);
|
||||
this.folderProgressMap.delete(id);
|
||||
this.lastProgressUpdate.delete(id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset all progress data
|
||||
*/
|
||||
resetAllProgress(): void {
|
||||
this.fileProgressMap.clear();
|
||||
this.folderProgressMap.clear();
|
||||
this.lastProgressUpdate.clear();
|
||||
|
||||
// Reset speed calculator for current peer
|
||||
// Note: SpeedCalculator doesn't have resetSpeed method, so we create a new instance
|
||||
this.speedCalculator = new SpeedCalculator();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get progress update frequency (for debugging)
|
||||
*/
|
||||
getUpdateFrequency(id: string): number {
|
||||
const lastUpdate = this.lastProgressUpdate.get(id);
|
||||
return lastUpdate ? Date.now() - lastUpdate : 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if progress should be throttled
|
||||
*/
|
||||
shouldThrottleProgress(id: string, isFolder: boolean = false): boolean {
|
||||
const now = Date.now();
|
||||
const lastUpdate = this.lastProgressUpdate.get(id) || 0;
|
||||
const threshold = isFolder ? 200 : 100; // Folders update less frequently
|
||||
return now - lastUpdate < threshold;
|
||||
}
|
||||
|
||||
/**
|
||||
* Force progress update (bypass throttling)
|
||||
*/
|
||||
forceProgressUpdate(id: string, progress: number): void {
|
||||
if (!this.progressCallback) return;
|
||||
|
||||
const speed = this.getCurrentSpeed();
|
||||
this.progressCallback(id, progress, speed);
|
||||
this.lastProgressUpdate.set(id, Date.now());
|
||||
|
||||
// Update internal maps
|
||||
if (this.fileProgressMap.has(id)) {
|
||||
this.fileProgressMap.set(id, progress);
|
||||
} else if (this.folderProgressMap.has(id)) {
|
||||
this.folderProgressMap.set(id, progress);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up resources
|
||||
*/
|
||||
cleanup(): void {
|
||||
this.resetAllProgress();
|
||||
this.progressCallback = null;
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_PROGRESS_LOGGING) {
|
||||
postLogToBackend("[DEBUG] 🧹 ProgressReporter cleaned up");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
/**
|
||||
* 🚀 Reception configuration management
|
||||
* Centralized configuration for file reception parameters
|
||||
*/
|
||||
|
||||
export class ReceptionConfig {
|
||||
// File size thresholds
|
||||
static readonly FILE_CONFIG = {
|
||||
LARGE_FILE_THRESHOLD: 1 * 1024 * 1024 * 1024, // 1GB - files larger than this will be saved to disk
|
||||
CHUNK_SIZE: 65536, // 64KB standard chunk size
|
||||
};
|
||||
|
||||
// Buffer management
|
||||
static readonly BUFFER_CONFIG = {
|
||||
MAX_BUFFER_SIZE: 100, // Buffer up to 100 chunks (approximately 6.4MB)
|
||||
SEQUENTIAL_FLUSH_THRESHOLD: 10, // Start flushing when this many sequential chunks are available
|
||||
};
|
||||
|
||||
// Performance and debugging
|
||||
static readonly DEBUG_CONFIG = {
|
||||
ENABLE_CHUNK_LOGGING: process.env.NODE_ENV === "development",
|
||||
ENABLE_PROGRESS_LOGGING: process.env.NODE_ENV === "development",
|
||||
PROGRESS_LOG_INTERVAL: 500, // Log progress every N chunks
|
||||
COMPLETION_CHECK_INTERVAL: 100, // Check completion every N ms
|
||||
};
|
||||
|
||||
// Network and timing
|
||||
static readonly NETWORK_CONFIG = {
|
||||
FIREFOX_COMPATIBILITY_DELAY: 10, // ms delay for Firefox compatibility
|
||||
FINALIZATION_TIMEOUT: 30000, // 30s timeout for file finalization
|
||||
GRACEFUL_SHUTDOWN_TIMEOUT: 5000, // 5s timeout for graceful shutdown
|
||||
};
|
||||
|
||||
// Validation thresholds
|
||||
static readonly VALIDATION_CONFIG = {
|
||||
MAX_SIZE_DIFFERENCE_BYTES: 1024, // Allow up to 1KB size difference for validation
|
||||
MIN_PACKET_SIZE: 4, // Minimum embedded packet size (4 bytes for length header)
|
||||
};
|
||||
|
||||
/**
|
||||
* Get chunk index from file offset
|
||||
*/
|
||||
static getChunkIndexFromOffset(offset: number): number {
|
||||
return Math.floor(offset / this.FILE_CONFIG.CHUNK_SIZE);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get file offset from chunk index
|
||||
*/
|
||||
static getOffsetFromChunkIndex(chunkIndex: number): number {
|
||||
return chunkIndex * this.FILE_CONFIG.CHUNK_SIZE;
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate expected chunks count for file size and offset
|
||||
*/
|
||||
static calculateExpectedChunks(fileSize: number, startOffset: number = 0): number {
|
||||
return Math.ceil((fileSize - startOffset) / this.FILE_CONFIG.CHUNK_SIZE);
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculate total chunks in file
|
||||
*/
|
||||
static calculateTotalChunks(fileSize: number): number {
|
||||
return Math.ceil(fileSize / this.FILE_CONFIG.CHUNK_SIZE);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if file should be saved to disk
|
||||
*/
|
||||
static shouldSaveToDisk(fileSize: number, hasSaveDirectory: boolean): boolean {
|
||||
return hasSaveDirectory || fileSize >= this.FILE_CONFIG.LARGE_FILE_THRESHOLD;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,358 @@
|
||||
import {
|
||||
fileMetadata,
|
||||
FolderProgress,
|
||||
CurrentString,
|
||||
CustomFile,
|
||||
} from "@/types/webrtc";
|
||||
|
||||
/**
|
||||
* 🚀 Active file reception state interface
|
||||
*/
|
||||
export interface ActiveFileReception {
|
||||
meta: fileMetadata;
|
||||
chunks: (ArrayBuffer | null)[];
|
||||
receivedSize: number;
|
||||
initialOffset: number;
|
||||
fileHandle: FileSystemFileHandle | null;
|
||||
writeStream: FileSystemWritableFileStream | null;
|
||||
sequencedWriter: any | null; // Will be typed properly when StreamingFileWriter is implemented
|
||||
completionNotifier: {
|
||||
resolve: () => void;
|
||||
reject: (reason?: any) => void;
|
||||
};
|
||||
receivedChunksCount: number;
|
||||
expectedChunksCount: number;
|
||||
chunkSequenceMap: Map<number, boolean>;
|
||||
isFinalized?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* 🚀 Reception state management
|
||||
* Centrally manages all file reception state data
|
||||
*/
|
||||
export class ReceptionStateManager {
|
||||
// File metadata management
|
||||
private pendingFilesMeta = new Map<string, fileMetadata>();
|
||||
|
||||
// Folder progress tracking
|
||||
private folderProgresses: Record<string, FolderProgress> = {};
|
||||
|
||||
// Save type configuration (fileId/folderName -> isSavedToDisk)
|
||||
public saveType: Record<string, boolean> = {};
|
||||
|
||||
// Active transfer states
|
||||
private activeFileReception: ActiveFileReception | null = null;
|
||||
private activeStringReception: CurrentString | null = null;
|
||||
private currentFolderName: string | null = null;
|
||||
|
||||
// Peer information
|
||||
private currentPeerId: string = "";
|
||||
private saveDirectory: FileSystemDirectoryHandle | null = null;
|
||||
|
||||
// ===== File Metadata Management =====
|
||||
|
||||
/**
|
||||
* Add file metadata
|
||||
*/
|
||||
public addFileMetadata(metadata: fileMetadata): boolean {
|
||||
if (this.pendingFilesMeta.has(metadata.fileId)) {
|
||||
return false; // Already exists
|
||||
}
|
||||
|
||||
this.pendingFilesMeta.set(metadata.fileId, metadata);
|
||||
|
||||
// Update folder progress if this file belongs to a folder
|
||||
if (metadata.folderName) {
|
||||
this.addFileToFolder(metadata.folderName, metadata.fileId, metadata.size);
|
||||
}
|
||||
|
||||
return true; // New metadata added
|
||||
}
|
||||
|
||||
/**
|
||||
* Get file metadata by ID
|
||||
*/
|
||||
public getFileMetadata(fileId: string): fileMetadata | undefined {
|
||||
return this.pendingFilesMeta.get(fileId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all pending file metadata
|
||||
*/
|
||||
public getAllFileMetadata(): Map<string, fileMetadata> {
|
||||
return new Map(this.pendingFilesMeta);
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove file metadata
|
||||
*/
|
||||
public removeFileMetadata(fileId: string): void {
|
||||
this.pendingFilesMeta.delete(fileId);
|
||||
}
|
||||
|
||||
// ===== Folder Progress Management =====
|
||||
|
||||
/**
|
||||
* Add file to folder progress tracking
|
||||
*/
|
||||
private addFileToFolder(folderName: string, fileId: string, fileSize: number): void {
|
||||
if (!this.folderProgresses[folderName]) {
|
||||
this.folderProgresses[folderName] = {
|
||||
totalSize: 0,
|
||||
receivedSize: 0,
|
||||
fileIds: [],
|
||||
};
|
||||
}
|
||||
|
||||
const folderProgress = this.folderProgresses[folderName];
|
||||
if (!folderProgress.fileIds.includes(fileId)) {
|
||||
folderProgress.fileIds.push(fileId);
|
||||
folderProgress.totalSize += fileSize;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get folder progress
|
||||
*/
|
||||
public getFolderProgress(folderName: string): FolderProgress | undefined {
|
||||
return this.folderProgresses[folderName];
|
||||
}
|
||||
|
||||
/**
|
||||
* Update folder received size
|
||||
*/
|
||||
public updateFolderReceivedSize(folderName: string, additionalBytes: number): void {
|
||||
const folderProgress = this.folderProgresses[folderName];
|
||||
if (folderProgress) {
|
||||
folderProgress.receivedSize += additionalBytes;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set folder received size (for resume scenarios)
|
||||
*/
|
||||
public setFolderReceivedSize(folderName: string, totalReceivedSize: number): void {
|
||||
const folderProgress = this.folderProgresses[folderName];
|
||||
if (folderProgress) {
|
||||
folderProgress.receivedSize = totalReceivedSize;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all folder progresses
|
||||
*/
|
||||
public getAllFolderProgresses(): Record<string, FolderProgress> {
|
||||
return { ...this.folderProgresses };
|
||||
}
|
||||
|
||||
// ===== Active File Reception Management =====
|
||||
|
||||
/**
|
||||
* Start active file reception
|
||||
*/
|
||||
public startFileReception(
|
||||
meta: fileMetadata,
|
||||
expectedChunksCount: number,
|
||||
initialOffset: number = 0
|
||||
): Promise<void> {
|
||||
if (this.activeFileReception) {
|
||||
throw new Error("Another file reception is already in progress");
|
||||
}
|
||||
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
this.activeFileReception = {
|
||||
meta,
|
||||
chunks: new Array(expectedChunksCount).fill(null),
|
||||
receivedSize: 0,
|
||||
initialOffset,
|
||||
fileHandle: null,
|
||||
writeStream: null,
|
||||
sequencedWriter: null,
|
||||
completionNotifier: { resolve, reject },
|
||||
receivedChunksCount: 0,
|
||||
expectedChunksCount,
|
||||
chunkSequenceMap: new Map<number, boolean>(),
|
||||
isFinalized: false,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get active file reception
|
||||
*/
|
||||
public getActiveFileReception(): ActiveFileReception | null {
|
||||
return this.activeFileReception;
|
||||
}
|
||||
|
||||
/**
|
||||
* Update active file reception
|
||||
*/
|
||||
public updateActiveFileReception(updates: Partial<ActiveFileReception>): void {
|
||||
if (this.activeFileReception) {
|
||||
Object.assign(this.activeFileReception, updates);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Complete active file reception
|
||||
*/
|
||||
public completeFileReception(): void {
|
||||
if (this.activeFileReception?.completionNotifier) {
|
||||
this.activeFileReception.completionNotifier.resolve();
|
||||
}
|
||||
this.activeFileReception = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fail active file reception
|
||||
*/
|
||||
public failFileReception(reason: any): void {
|
||||
if (this.activeFileReception?.completionNotifier) {
|
||||
this.activeFileReception.completionNotifier.reject(reason);
|
||||
}
|
||||
this.activeFileReception = null;
|
||||
}
|
||||
|
||||
// ===== String Reception Management =====
|
||||
|
||||
/**
|
||||
* Start string reception
|
||||
*/
|
||||
public startStringReception(length: number): void {
|
||||
this.activeStringReception = {
|
||||
length,
|
||||
chunks: [],
|
||||
receivedChunks: 0,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get active string reception
|
||||
*/
|
||||
public getActiveStringReception(): CurrentString | null {
|
||||
return this.activeStringReception;
|
||||
}
|
||||
|
||||
/**
|
||||
* Update string reception chunk
|
||||
*/
|
||||
public updateStringReceptionChunk(index: number, chunk: string): void {
|
||||
if (this.activeStringReception) {
|
||||
this.activeStringReception.chunks[index] = chunk;
|
||||
this.activeStringReception.receivedChunks++;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Complete string reception
|
||||
*/
|
||||
public completeStringReception(): string | null {
|
||||
if (!this.activeStringReception) return null;
|
||||
|
||||
const fullString = this.activeStringReception.chunks.join("");
|
||||
this.activeStringReception = null;
|
||||
return fullString;
|
||||
}
|
||||
|
||||
// ===== Current Context Management =====
|
||||
|
||||
/**
|
||||
* Set current folder name
|
||||
*/
|
||||
public setCurrentFolderName(folderName: string | null): void {
|
||||
this.currentFolderName = folderName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current folder name
|
||||
*/
|
||||
public getCurrentFolderName(): string | null {
|
||||
return this.currentFolderName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set current peer ID
|
||||
*/
|
||||
public setCurrentPeerId(peerId: string): void {
|
||||
this.currentPeerId = peerId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get current peer ID
|
||||
*/
|
||||
public getCurrentPeerId(): string {
|
||||
return this.currentPeerId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set save directory
|
||||
*/
|
||||
public setSaveDirectory(directory: FileSystemDirectoryHandle | null): void {
|
||||
this.saveDirectory = directory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get save directory
|
||||
*/
|
||||
public getSaveDirectory(): FileSystemDirectoryHandle | null {
|
||||
return this.saveDirectory;
|
||||
}
|
||||
|
||||
// ===== Save Type Management =====
|
||||
|
||||
/**
|
||||
* Set save type for file or folder
|
||||
*/
|
||||
public setSaveType(id: string, saveToDisk: boolean): void {
|
||||
this.saveType[id] = saveToDisk;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get save type for file or folder
|
||||
*/
|
||||
public getSaveType(id: string): boolean {
|
||||
return this.saveType[id] || false;
|
||||
}
|
||||
|
||||
// ===== State Reset and Cleanup =====
|
||||
|
||||
/**
|
||||
* Force reset all states (for reconnection scenarios)
|
||||
*/
|
||||
public forceReset(): void {
|
||||
this.pendingFilesMeta.clear();
|
||||
this.folderProgresses = {};
|
||||
this.saveType = {};
|
||||
this.activeFileReception = null;
|
||||
this.activeStringReception = null;
|
||||
this.currentFolderName = null;
|
||||
this.currentPeerId = "";
|
||||
// Note: saveDirectory is preserved
|
||||
}
|
||||
|
||||
/**
|
||||
* Graceful cleanup (preserve some state for potential resume)
|
||||
*/
|
||||
public gracefulCleanup(): void {
|
||||
this.activeFileReception = null;
|
||||
this.activeStringReception = null;
|
||||
this.currentFolderName = null;
|
||||
// Note: preserve pendingFilesMeta, folderProgresses, saveType for potential resume
|
||||
}
|
||||
|
||||
/**
|
||||
* Get state statistics (for debugging)
|
||||
*/
|
||||
public getStateStats() {
|
||||
return {
|
||||
pendingFilesCount: this.pendingFilesMeta.size,
|
||||
folderCount: Object.keys(this.folderProgresses).length,
|
||||
hasActiveFileReception: !!this.activeFileReception,
|
||||
hasActiveStringReception: !!this.activeStringReception,
|
||||
currentFolderName: this.currentFolderName,
|
||||
currentPeerId: this.currentPeerId,
|
||||
hasSaveDirectory: !!this.saveDirectory,
|
||||
saveTypeCount: Object.keys(this.saveType).length,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,398 @@
|
||||
import { ReceptionConfig } from "./ReceptionConfig";
|
||||
import { postLogToBackend } from "@/app/config/api";
|
||||
|
||||
const developmentEnv = process.env.NODE_ENV;
|
||||
|
||||
/**
|
||||
* 🚀 Strict Sequential Buffering Writer - Optimizes large file disk I/O performance
|
||||
*/
|
||||
export class SequencedDiskWriter {
|
||||
private writeQueue = new Map<number, ArrayBuffer>();
|
||||
private nextWriteIndex = 0;
|
||||
private readonly maxBufferSize: number;
|
||||
private readonly stream: FileSystemWritableFileStream;
|
||||
private totalWritten = 0;
|
||||
|
||||
constructor(stream: FileSystemWritableFileStream, startIndex: number = 0) {
|
||||
this.stream = stream;
|
||||
this.nextWriteIndex = startIndex;
|
||||
this.maxBufferSize = ReceptionConfig.BUFFER_CONFIG.MAX_BUFFER_SIZE;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write a chunk, automatically managing order and buffering
|
||||
*/
|
||||
async writeChunk(chunkIndex: number, chunk: ArrayBuffer): Promise<void> {
|
||||
// Debug writeChunk calls
|
||||
if (
|
||||
ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING &&
|
||||
(chunkIndex <= 5 || chunkIndex === this.nextWriteIndex)
|
||||
) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-RESUME] 🎯 WriteChunk called - received:${chunkIndex}, expected:${
|
||||
this.nextWriteIndex
|
||||
}, match:${chunkIndex === this.nextWriteIndex}`
|
||||
);
|
||||
}
|
||||
|
||||
// 1. If it is the expected next chunk, write immediately
|
||||
if (chunkIndex === this.nextWriteIndex) {
|
||||
await this.flushSequentialChunks(chunk);
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. If it's a future chunk, buffer it
|
||||
if (chunkIndex > this.nextWriteIndex) {
|
||||
if (this.writeQueue.size < this.maxBufferSize) {
|
||||
this.writeQueue.set(chunkIndex, chunk);
|
||||
} else {
|
||||
// Buffer full, forcing processing of the earliest chunk to free up space
|
||||
await this.forceFlushOldest();
|
||||
this.writeQueue.set(chunkIndex, chunk);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. If the chunk is expired, log a warning but ignore (already written)
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ⚠️ DUPLICATE chunk #${chunkIndex} ignored (already written #${this.nextWriteIndex})`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Write current chunk and attempt to sequentially write subsequent chunks
|
||||
*/
|
||||
private async flushSequentialChunks(firstChunk: ArrayBuffer): Promise<void> {
|
||||
let flushCount = 0;
|
||||
|
||||
try {
|
||||
// Write current chunk
|
||||
await this.stream.write(firstChunk);
|
||||
this.totalWritten += firstChunk.byteLength;
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] ✓ DISK_WRITE chunk #${this.nextWriteIndex} - ${firstChunk.byteLength} bytes, total: ${this.totalWritten}`
|
||||
);
|
||||
}
|
||||
|
||||
this.nextWriteIndex++;
|
||||
|
||||
// Try to sequentially write chunks from buffer
|
||||
while (this.writeQueue.has(this.nextWriteIndex)) {
|
||||
const chunk = this.writeQueue.get(this.nextWriteIndex)!;
|
||||
await this.stream.write(chunk);
|
||||
this.totalWritten += chunk.byteLength;
|
||||
this.writeQueue.delete(this.nextWriteIndex);
|
||||
|
||||
flushCount++;
|
||||
this.nextWriteIndex++;
|
||||
}
|
||||
} catch (error) {
|
||||
// Defensive handling: If stream is closed, silently ignore
|
||||
const errorMessage =
|
||||
error instanceof Error ? error.message : String(error);
|
||||
if (
|
||||
errorMessage.includes("closing writable stream") ||
|
||||
errorMessage.includes("stream is closed")
|
||||
) {
|
||||
console.log(
|
||||
`[SequencedDiskWriter] Stream closed during write - ignoring remaining chunks`
|
||||
);
|
||||
return;
|
||||
}
|
||||
// Re-throw other types of errors
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (flushCount > 0 && ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 🔥 SEQUENTIAL_FLUSH ${flushCount} chunks, now at #${this.nextWriteIndex}, queue: ${this.writeQueue.size}`
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the next expected write index
|
||||
*/
|
||||
get expectedIndex(): number {
|
||||
return this.nextWriteIndex;
|
||||
}
|
||||
|
||||
/**
|
||||
* Force flush the earliest chunk to release buffer space
|
||||
*/
|
||||
private async forceFlushOldest(): Promise<void> {
|
||||
try {
|
||||
if (this.writeQueue.size === 0) return;
|
||||
|
||||
const oldestIndex = Math.min(...Array.from(this.writeQueue.keys()));
|
||||
const chunk = this.writeQueue.get(oldestIndex)!;
|
||||
|
||||
// Use seek to write at the correct position (fallback handling)
|
||||
const fileOffset = ReceptionConfig.getOffsetFromChunkIndex(oldestIndex);
|
||||
await this.stream.seek(fileOffset);
|
||||
await this.stream.write(chunk);
|
||||
this.writeQueue.delete(oldestIndex);
|
||||
|
||||
// Return to current position
|
||||
const currentOffset = ReceptionConfig.getOffsetFromChunkIndex(this.nextWriteIndex);
|
||||
await this.stream.seek(currentOffset);
|
||||
} catch (error) {
|
||||
// Defensive handling: If stream is closed, silently ignore
|
||||
const errorMessage =
|
||||
error instanceof Error ? error.message : String(error);
|
||||
if (
|
||||
errorMessage.includes("closing writable stream") ||
|
||||
errorMessage.includes("stream is closed")
|
||||
) {
|
||||
console.log(
|
||||
`[SequencedDiskWriter] Stream closed during write - ignoring remaining chunks`
|
||||
);
|
||||
return;
|
||||
}
|
||||
// Re-throw other types of errors
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get buffer status
|
||||
*/
|
||||
getBufferStatus(): {
|
||||
queueSize: number;
|
||||
nextIndex: number;
|
||||
totalWritten: number;
|
||||
} {
|
||||
return {
|
||||
queueSize: this.writeQueue.size,
|
||||
nextIndex: this.nextWriteIndex,
|
||||
totalWritten: this.totalWritten,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Close and clean up resources
|
||||
*/
|
||||
async close(): Promise<void> {
|
||||
try {
|
||||
// Try to flush all remaining chunks
|
||||
const remainingIndexes = Array.from(this.writeQueue.keys()).sort(
|
||||
(a, b) => a - b
|
||||
);
|
||||
for (const chunkIndex of remainingIndexes) {
|
||||
const chunk = this.writeQueue.get(chunkIndex)!;
|
||||
const fileOffset = ReceptionConfig.getOffsetFromChunkIndex(chunkIndex);
|
||||
await this.stream.seek(fileOffset);
|
||||
await this.stream.write(chunk);
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 💾 FINAL_FLUSH chunk #${chunkIndex} at cleanup`
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// Defensive handling: If stream is not writable during close, handle silently
|
||||
const errorMessage =
|
||||
error instanceof Error ? error.message : String(error);
|
||||
if (
|
||||
errorMessage.includes("closing writable stream") ||
|
||||
errorMessage.includes("stream is closed")
|
||||
) {
|
||||
console.log(
|
||||
`[SequencedDiskWriter] Stream closed during final flush - data may be incomplete`
|
||||
);
|
||||
} else {
|
||||
console.warn(
|
||||
`[SequencedDiskWriter] Error during final flush:`,
|
||||
errorMessage
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
this.writeQueue.clear();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 🚀 Streaming file writer
|
||||
* Manages disk file creation, directory structure, and streaming writes
|
||||
*/
|
||||
export class StreamingFileWriter {
|
||||
private saveDirectory: FileSystemDirectoryHandle | null = null;
|
||||
|
||||
constructor(saveDirectory?: FileSystemDirectoryHandle) {
|
||||
this.saveDirectory = saveDirectory || null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set save directory
|
||||
*/
|
||||
setSaveDirectory(directory: FileSystemDirectoryHandle): void {
|
||||
this.saveDirectory = directory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create disk write stream for a file
|
||||
*/
|
||||
async createWriteStream(
|
||||
fileName: string,
|
||||
fullPath: string,
|
||||
offset: number = 0
|
||||
): Promise<{
|
||||
fileHandle: FileSystemFileHandle;
|
||||
writeStream: FileSystemWritableFileStream;
|
||||
sequencedWriter: SequencedDiskWriter;
|
||||
}> {
|
||||
if (!this.saveDirectory) {
|
||||
throw new Error("Save directory not set");
|
||||
}
|
||||
|
||||
try {
|
||||
const folderHandle = await this.createFolderStructure(fullPath);
|
||||
const fileHandle = await folderHandle.getFileHandle(fileName, {
|
||||
create: true,
|
||||
});
|
||||
|
||||
// Use keepExistingData: true to append
|
||||
const writeStream = await fileHandle.createWritable({
|
||||
keepExistingData: true,
|
||||
});
|
||||
|
||||
// Seek to the offset to start writing from there
|
||||
await writeStream.seek(offset);
|
||||
|
||||
// Create strictly sequential write manager
|
||||
const startChunkIndex = ReceptionConfig.getChunkIndexFromOffset(offset);
|
||||
const sequencedWriter = new SequencedDiskWriter(writeStream, startChunkIndex);
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG] 📢 SEQUENCED_WRITER created - startIndex: ${startChunkIndex}, offset: ${offset}`
|
||||
);
|
||||
postLogToBackend(
|
||||
`[DEBUG-RESUME] 🎯 SequencedWriter expects - startIndex:${startChunkIndex}, offset:${offset}, calculatedFrom:${offset}/65536`
|
||||
);
|
||||
}
|
||||
|
||||
return { fileHandle, writeStream, sequencedWriter };
|
||||
} catch (err) {
|
||||
throw new Error(`Failed to create file on disk: ${err}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if partial file exists and get its size
|
||||
*/
|
||||
async getPartialFileSize(fileName: string, fullPath: string): Promise<number> {
|
||||
if (!this.saveDirectory) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
try {
|
||||
const folderHandle = await this.createFolderStructure(fullPath);
|
||||
const fileHandle = await folderHandle.getFileHandle(fileName, {
|
||||
create: false,
|
||||
});
|
||||
const file = await fileHandle.getFile();
|
||||
return file.size;
|
||||
} catch {
|
||||
// File does not exist
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create folder structure based on full path
|
||||
*/
|
||||
private async createFolderStructure(
|
||||
fullPath: string
|
||||
): Promise<FileSystemDirectoryHandle> {
|
||||
if (!this.saveDirectory) {
|
||||
throw new Error("Save directory not set");
|
||||
}
|
||||
|
||||
const parts = fullPath.split("/");
|
||||
parts.pop(); // Remove filename
|
||||
|
||||
let currentDir = this.saveDirectory;
|
||||
for (const part of parts) {
|
||||
if (part) {
|
||||
currentDir = await currentDir.getDirectoryHandle(part, {
|
||||
create: true,
|
||||
});
|
||||
}
|
||||
}
|
||||
return currentDir;
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalize file write and close streams
|
||||
*/
|
||||
async finalizeWrite(
|
||||
sequencedWriter: SequencedDiskWriter,
|
||||
writeStream: FileSystemWritableFileStream,
|
||||
fileName: string
|
||||
): Promise<void> {
|
||||
try {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] 🚀 Starting finalization for ${fileName}`
|
||||
);
|
||||
}
|
||||
|
||||
// First close the strict sequential writing manager (flush all buffers)
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(`[DEBUG-FINALIZE] Closing SequencedWriter...`);
|
||||
}
|
||||
await sequencedWriter.close();
|
||||
const status = sequencedWriter.getBufferStatus();
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] 💾 SEQUENCED_WRITER closed - totalWritten: ${status.totalWritten}, finalQueue: ${status.queueSize}`
|
||||
);
|
||||
}
|
||||
|
||||
// Then close the file stream
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] About to close writeStream for ${fileName}`
|
||||
);
|
||||
}
|
||||
await writeStream.close();
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(`[DEBUG-FINALIZE] ✅ WriteStream closed successfully`);
|
||||
}
|
||||
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] ✅ LARGE_FILE finalized successfully - ${fileName}`
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
if (ReceptionConfig.DEBUG_CONFIG.ENABLE_CHUNK_LOGGING) {
|
||||
postLogToBackend(
|
||||
`[DEBUG-FINALIZE] ❌ Error during finalization: ${error}`
|
||||
);
|
||||
}
|
||||
throw new Error(`Error finalizing large file: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if save directory is available
|
||||
*/
|
||||
hasSaveDirectory(): boolean {
|
||||
return !!this.saveDirectory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get save directory
|
||||
*/
|
||||
getSaveDirectory(): FileSystemDirectoryHandle | null {
|
||||
return this.saveDirectory;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
/**
|
||||
* 🚀 File receive module unified export
|
||||
* Provides modular file reception services
|
||||
*/
|
||||
|
||||
// Configuration management
|
||||
export { ReceptionConfig } from "./ReceptionConfig";
|
||||
|
||||
// State management
|
||||
export { ReceptionStateManager } from "./ReceptionStateManager";
|
||||
export type { ActiveFileReception } from "./ReceptionStateManager";
|
||||
|
||||
// Data processing
|
||||
export { ChunkProcessor } from "./ChunkProcessor";
|
||||
export type { ChunkProcessingResult } from "./ChunkProcessor";
|
||||
|
||||
// File writing
|
||||
export { StreamingFileWriter, SequencedDiskWriter } from "./StreamingFileWriter";
|
||||
|
||||
// File assembly
|
||||
export { FileAssembler } from "./FileAssembler";
|
||||
export type { FileAssemblyResult } from "./FileAssembler";
|
||||
|
||||
// Message processing
|
||||
export { MessageProcessor } from "./MessageProcessor";
|
||||
export type { MessageProcessorDelegate } from "./MessageProcessor";
|
||||
|
||||
// Progress reporting
|
||||
export { ProgressReporter } from "./ProgressReporter";
|
||||
export type { ProgressCallback, ProgressStats } from "./ProgressReporter";
|
||||
|
||||
// Main orchestrator
|
||||
export { FileReceiveOrchestrator } from "./FileReceiveOrchestrator";
|
||||
|
||||
/**
|
||||
* 🎯 Convenience creation function - Quick initialization of file receive services
|
||||
*/
|
||||
import WebRTC_Recipient from "../webrtc_Recipient";
|
||||
import { FileReceiveOrchestrator } from "./FileReceiveOrchestrator";
|
||||
|
||||
export function createFileReceiveService(
|
||||
webrtcConnection: WebRTC_Recipient
|
||||
): FileReceiveOrchestrator {
|
||||
return new FileReceiveOrchestrator(webrtcConnection);
|
||||
}
|
||||
Reference in New Issue
Block a user