Function streamToNdStrings

app/src/utils/azure/server.ts:91–151

({
  readableStream,
  maxEntriesToImport,
  onProgress,
  chunkInterval = 1048576, // send progress every 1MB
}: {
  readableStream: NodeJS.ReadableStream;
  maxEntriesToImport: number;
  onProgress?: (progress: number) => Promise<void>;
  chunkInterval?: number;
})

// Splits the stream into individual chunks split on newlines
async function streamToNdStrings({
  readableStream,
  maxEntriesToImport,
  onProgress,
  chunkInterval = 1048576, // send progress every 1MB
}: {
  readableStream: NodeJS.ReadableStream;
  maxEntriesToImport: number;
  onProgress?: (progress: number) => Promise<void>;
  chunkInterval?: number;
}): Promise<string[]> {
  return new Promise((resolve, reject) => {
    const lines: string[] = [];
    let bytesDownloaded = 0;
    let lastReportedByteCount = 0;
    let tempBuffer: Buffer = Buffer.alloc(0);
    let numEntriesImported = 0;

    readableStream.on("data", (chunk: Buffer) => {
      bytesDownloaded += chunk.byteLength;

      // Report progress
      if (onProgress && bytesDownloaded - lastReportedByteCount >= chunkInterval) {
        void onProgress(bytesDownloaded);
        lastReportedByteCount = bytesDownloaded;
      }

      // Combine with leftover buffer from previous chunk
      chunk = Buffer.concat([tempBuffer, chunk]);

      let newlineIndex;
      while (
        (newlineIndex = chunk.indexOf(0x0a)) !== -1 &&
        numEntriesImported < maxEntriesToImport
      ) {
        const line = chunk.slice(0, newlineIndex).toString("utf-8");
        lines.push(line);
        chunk = chunk.slice(newlineIndex + 1);
        numEntriesImported++;
      }

      if (numEntriesImported >= maxEntriesToImport) {
        // TODO: cancel the stream
        resolve(lines);
        return;
      }

      // Save leftover data for next chunk
      tempBuffer = chunk;
    });

    readableStream.on("end", () => {
      if (tempBuffer.length > 0) {
        lines.push(tempBuffer.toString("utf-8")); // add the last part
      }
      resolve(lines);
    });