gzipConcatenation.mjs

import zlib from "node:zlib";
import http from "node:http";
import { inspect } from "node:util";
import { PassThrough } from "node:stream";

// crc32

const GF2_DIM = 32;

function gf2MatrixTimes(mat, vec) {
  let sum = 0;
  let i = 0;

  while (vec !== 0) {
    if (vec & (1 !== 0)) {
      sum = (sum ^ mat[i]) >>> 0;
    }
    vec >>>= 1;
    ++i;
  }
  return sum;
}

function gf2MatrixSquare(square, mat) {
  for (let i = 0; i < GF2_DIM; ++i) {
    square[i] = gf2MatrixTimes(mat, mat[i]);
  }
}

function combineCrc32(poly, crc1, crc2, len2) {
  // degenerate case (also disallow negative lengths)
  if (len2 <= 0) {
    return crc1;
  }

  const even = Array(GF2_DIM);
  const odd = Array(GF2_DIM);

  odd[0] = poly; // CRC-32 polynomial
  let row = 1;
  for (let i = 1; i < GF2_DIM; i++) {
    odd[i] = row;
    row <<= 1;
  }

  // put operator for two zero bits in even
  gf2MatrixSquare(even, odd);

  // put operator for four zero bits in odd
  gf2MatrixSquare(odd, even);

  // apply len2 zeros to crc1 (first square will put the operator for one
  // zero byte, eight zero bits, in even)
  let crc1n = crc1;
  while (true) {
    // apply zeros operator for this bit of len2
    gf2MatrixSquare(even, odd);
    if (len2 & (1 != 0)) {
      crc1n = gf2MatrixTimes(even, crc1n);
    }
    len2 >>>= 1;

    // if no more bits set, then done
    if (len2 == 0) {
      break;
    }

    // another iteration of the loop with odd and even swapped
    gf2MatrixSquare(odd, even);
    if (len2 & (1 != 0)) {
      crc1n = gf2MatrixTimes(odd, crc1n);
    }
    len2 >>>= 1;

    // if no more bits set, then done
    if (len2 == 0) {
      break;
    }
  }

  // return combined crc
  crc1n = (crc1n ^ crc2) >>> 0;
  return crc1n;
}

// adler32

function combineAdler32(adler1, adler2, len2) {
  const BASE = 65521;
  const WORD = 0xffff;
  const DWORD = 0xffffffff;
  if (adler1 < 0 || adler1 > DWORD) {
    throw new Error("bad adler1");
  }
  if (adler2 < 0 || adler2 > DWORD) {
    throw new Error("bad adler2");
  }
  if (len2 < 0) {
    throw new Error("bad len2");
  }

  const remainder = len2 % BASE;
  let sum1 = adler1 & WORD;
  let sum2 = (remainder * sum1) % BASE;
  sum1 += (adler2 & WORD) + BASE - 1;
  sum2 += ((adler1 >>> 16) & WORD) + ((adler2 >>> 16) & WORD) + BASE - remainder;
  if (sum1 >= BASE) {
    sum1 -= BASE;
  }
  if (sum1 >= BASE) {
    sum1 -= BASE;
  }
  if (sum2 >= BASE << 1) {
    sum2 -= BASE << 1;
  }
  if (sum2 >= BASE) {
    sum2 -= BASE;
  }
  return (sum1 | (sum2 << 16)) >>> 0;
}

async function streamToBuffer(stream, fn) {
  const streamPromise = new Promise((resolve, reject) => {
    const _buf = [];
    stream.on("data", (chunk) => _buf.push(chunk));
    stream.on("error", (err) => reject(err));
    stream.on("end", () => resolve(Buffer.concat(_buf)));
  });

  return await Promise.all([streamPromise, fn(stream)]).then(([buf]) => buf);
}

const DEFLATE_HEADER = Buffer.from([0x78, 0x9c]);
// 0x78 -- compression method: deflate; compression info: 32k window
// 0x9c -- 10 (compressor default level) 0 (no dict) 11100 (check bits)
const DEFLATE_EMPTY_BLOCK = Buffer.from([0x03, 0x00]);
const GZIP_HEADER = Buffer.from([
  0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
]);
const GZIP_CRC32_POLY = 0xedb88320; // In reversed form; 0x04c11db7 non-reversed

/** Writes a buffer to a stream, as a promise */
async function writeStream(stream, buffer) {
  return new Promise((resolve, reject) =>
    stream.write(buffer, (err) => (err ? reject(err) : resolve())),
  );
}

/** Creates a payload (with all requisite metadata) that can be merged into a deflate or gzip stream */
async function createDeflatePayload(buffer) {
  const deflated = await streamToBuffer(zlib.createDeflate({}), async (deflateStream) => {
    await writeStream(deflateStream, buffer);
    // I think this MUST be awaited or the `end` can screw up the stream
    await new Promise((resolve) =>
      deflateStream.flush(zlib.constants.Z_FULL_FLUSH, resolve),
    );
    await new Promise((resolve) => deflateStream.end(resolve));
  });
  return {
    // Payload is without the deflate header, "empty block" trailer, or adler32
    // Header: 0x78 0x9c 
    // Payload: <N bytes>
    // Empty block trailer: 0x03 0x00 
    // Adler32: <4 bytes>
    payload: deflated.subarray(2, -6), 
    adler32: deflated.subarray(-4).readUint32BE(),
    length: buffer.length,
    crc32: zlib.crc32(buffer),
  };
}

/** Concatenates 1 or more payloads into a gzip stream, writes to writable */
async function concatenateToGzip(payloads, writable) {
  await writeStream(writable, GZIP_HEADER);
  let totalLength = 0;
  let finalCrc32 = 0;
  for (let i = 0; i < payloads.length; i++) {
    const payload = payloads[i];
    totalLength += payload.length;
    await writeStream(writable, payload.payload);
    if (i === 0) {
      finalCrc32 = payload.crc32;
    } else {
      finalCrc32 = combineCrc32(
        GZIP_CRC32_POLY,
        finalCrc32,
        payload.crc32,
        payload.length,
      );
    }
  }
  await writeStream(writable, DEFLATE_EMPTY_BLOCK);
  const lengthMod32 = totalLength & 0xffffffff;
  const trailer = Buffer.alloc(8);
  trailer.writeUInt32LE(finalCrc32, 0);
  trailer.writeUInt32LE(lengthMod32, 4);
  await writeStream(writable, trailer);
}

/** Concatenates 1 or more payloads into a raw deflate stream, writes to writable */
async function concatenateToDeflate(payloads, writable) {
  await writeStream(writable, DEFLATE_HEADER);
  let totalLength = 0;
  let finalAdler32 = 0;
  for (let i = 0; i < payloads.length; i++) {
    const payload = payloads[i];
    totalLength += payload.length;
    await writeStream(writable, payload.payload);
    if (i === 0) {
      finalAdler32 = payload.adler32;
    } else {
      finalAdler32 = combineAdler32(
        finalAdler32,
        payload.adler32,
        payload.length,
      );
    }
  }
  await writeStream(writable, DEFLATE_EMPTY_BLOCK);
  const adlerBuf = Buffer.alloc(4);
  adlerBuf.writeUInt32BE(finalAdler32);
  await writeStream(writable, adlerBuf);
}

// TESTS

const makeBuffer = (maxBufferLength = 20000) => {
    const payloadLength = Math.floor(Math.random() * maxBufferLength);
    const buffer = Buffer.alloc(payloadLength);
    for (let i = 0; i < payloadLength; ++i) {
        buffer[i] = Math.floor(Math.random() * 256);
    }
    return buffer
}

const makeBuffers = (maxBuffers = 100, maxBufferLength = 20000) => {
    const bufferCount = Math.floor(Math.random() * (maxBuffers - 1)) + 1;
    const buffers = Array(bufferCount);
    for (let i = 0; i < bufferCount; ++i) {
        buffers[i] = makeBuffer(maxBufferLength);
    }
    return buffers;
}

async function runTest() {
    const buffers = makeBuffers()
    const payloads = await Promise.all(buffers.map(createDeflatePayload));
    const mergedDeflate = await streamToBuffer(new PassThrough(), async (stream) => {
        await concatenateToDeflate(payloads, stream);
        await new Promise((resolve) => stream.end(resolve));
    })
    const mergedGzip = await streamToBuffer(new PassThrough(), async (stream) => {
        await concatenateToGzip(payloads, stream);
        await new Promise((resolve) => stream.end(resolve));
    })

    const expectedOutput = Buffer.concat(buffers);
    // Try decompressing and checking output:
    const decompressedDeflate = await new Promise((resolve, reject) => zlib.inflate(mergedDeflate, (err, res) => err ? reject(err) : resolve(res)));
    const decompressedGzip = await new Promise((resolve, reject) => zlib.gunzip(mergedGzip, (err, res) => err ? reject(err) : resolve(res)));

    let error = false;
    if (!decompressedDeflate.equals(expectedOutput)) {
        console.error("Deflate output does not match expected output");
        error =true;
    }
    if (!decompressedGzip.equals(expectedOutput)) {
        console.error("Deflate output does not match expected output");
        error = true;
    }
    if (error) {
        throw new Error("oh no")
    }
}

async function runTests() {
    for (let i = 0; i < 1000; ++i) {
        process.stdout.write(".")
        await runTest()
        process.stdout.write(":")
    }
}

await runTests().catch(err => console.log(err))

//await concatenateToGzip(await makePayloads(), process.stdout);

/*
const server = http.createServer(async (req, res) => {

  // Set headers to indicate streaming content
  res.writeHead(200, {
    'Content-Type': 'text/plain',
    'content-encoding': 'gzip',
  });

  await concatenateToGzip(await makePayloads(), res)

  res.end();
});

const PORT = 3000;
server.listen(PORT, () => {
  console.log(`Server running on port ${PORT}`);
});
*/