Function makeZlibTransform

lib/internal/streams/iter/transform.js:306–522

(createHandleFn, processFlag, finishFlag)

// so I/O and upstream work overlap with threadpool compression.
// ---------------------------------------------------------------------------
function makeZlibTransform(createHandleFn, processFlag, finishFlag) {
  return {
    __proto__: null,
    [kValidatedTransform]: true,
    transform: async function*(source, options) {
      const { signal } = options;

      // Fail fast if already aborted - don't allocate a native handle.
      signal?.throwIfAborted();

      // ---- Per-invocation state shared with the write callback ----
      let outBuf;
      let outOffset = 0;
      let chunkSize;
      let pending = [];
      let pendingBytes = 0;

      // Current write operation state (read by the callback for looping).
      let resolveWrite, rejectWrite;
      let writeInput, writeFlush;
      let writeInOff, writeAvailIn, writeAvailOutBefore;

      // processCallback: called by C++ AfterThreadPoolWork when compression
      // on the threadpool completes. Collects output, loops if the engine
      // has more output to produce (availOut === 0), then resolves the
      // promise when all output for this input chunk is collected.
      function onWriteComplete() {
        const availOut = writeState[0];
        const availInAfter = writeState[1];
        const have = writeAvailOutBefore - availOut;
        const bufferExhausted = availOut === 0 || outOffset + have >= chunkSize;

        if (have > 0) {
          if (bufferExhausted && outOffset === 0) {
            // Entire buffer filled from start - yield directly, no copy.
            ArrayPrototypePush(pending, outBuf);
          } else if (bufferExhausted) {
            // Tail of buffer filled and buffer is being replaced -
            // subarray is safe since outBuf reference is overwritten below.
            ArrayPrototypePush(pending,
                               outBuf.subarray(outOffset, outOffset + have));
          } else {
            // Partial fill, buffer will be reused - must copy.
            ArrayPrototypePush(pending,
                               TypedArrayPrototypeSlice(outBuf,
                                                        outOffset,
                                                        outOffset + have));
          }
          pendingBytes += have;
          outOffset += have;
        }

        // Reallocate output buffer if exhausted.
        if (bufferExhausted) {
          outBuf = Buffer.allocUnsafe(chunkSize);
          outOffset = 0;
        }