Parse JSON Lines Records Incrementally

export async function* parseJsonLines(source, validate) {
  let buffer = "";
  let lineNo = 0;
  for await (const chunk of source) {
    buffer += typeof chunk === "string" ? chunk : Buffer.from(chunk).toString("utf8");
    let nl;
    while ((nl = buffer.indexOf("\n")) !== -1) {
      const raw = buffer.slice(0, nl).replace(/\r$/, "");
      buffer = buffer.slice(nl + 1);
      lineNo += 1;
      if (raw.trim() === "") continue;
      let record;
      try {
        record = JSON.parse(raw);
      } catch (err) {
        throw new Error(`invalid JSON on line ${lineNo}: ${err.message}`);
      }
      if (validate) {
        const problem = validate(record);
        if (problem) throw new Error(`record failed validation on line ${lineNo}: ${problem}`);
      }
      yield { line: lineNo, record };
    }
  }
  if (buffer.trim() !== "") {
    lineNo += 1;
    let record;
    try {
      record = JSON.parse(buffer);
    } catch (err) {
      throw new Error(`invalid JSON on line ${lineNo}: ${err.message}`);
    }
    if (validate) {
      const problem = validate(record);
      if (problem) throw new Error(`record failed validation on line ${lineNo}: ${problem}`);
    }
    yield { line: lineNo, record };
  }
}