| 2383 | } |
| 2384 | // Process write queue with backpressure handling |
| 2385 | private async processQueue(): Promise<void> { |
| 2386 | if (this.isWriting || this.writeQueue.length === 0) { |
| 2387 | return; |
| 2388 | } |
| 2389 | this.isWriting = true; |
| 2390 | try { |
| 2391 | while (this.writeQueue.length > 0) { |
| 2392 | const batchBuffers: Buffer[] = []; |
| 2393 | let batchSize = 0; |
| 2394 | // Collect buffers up to batch size |
| 2395 | while (this.writeQueue.length > 0 && batchSize < this.batchSize) { |
| 2396 | const buffer = this.writeQueue.shift()!; |
| 2397 | batchBuffers.push(buffer); |
| 2398 | batchSize += buffer.length; |
| 2399 | this.queuedSize -= buffer.length; |
| 2400 | } |
| 2401 | if (batchBuffers.length > 0) { |
| 2402 | const combinedBuffer = Buffer.concat(batchBuffers); |
| 2403 | await this.writeToStream(combinedBuffer); |
| 2404 | } |
| 2405 | } |
| 2406 | } finally { |
| 2407 | this.isWriting = false; |
| 2408 | // Resolve any pending promises |
| 2409 | this.writePromises.forEach(({ resolve }) => resolve()); |
| 2410 | this.writePromises = []; |
| 2411 | } |
| 2412 | } |
| 2413 | // Write buffer to stream with backpressure handling |
| 2414 | private writeToStream(buffer: Buffer): Promise<void> { |
| 2415 | return new Promise((resolve, reject) => { |