Streaming NDJSON With Async Iteration

async function* ndjsonLines(body) {
  let buffer = '';
  const decoder = new TextDecoder();

  for await (const chunk of body) {
    buffer += decoder.decode(chunk, { stream: true });
    let newline;
    while ((newline = buffer.indexOf('\n')) !== -1) {
      const line = buffer.slice(0, newline).trim();
      buffer = buffer.slice(newline + 1);
      if (line) yield line;
    }
  }
  buffer += decoder.decode();
  if (buffer.trim()) yield buffer.trim();
}

async function ingest(url) {
  let processed = 0;
  let failed = 0;

  const res = await fetch(url, { headers: { Accept: 'application/x-ndjson' } });
  if (!res.ok) throw new Error('stream failed: ' + res.status);

  for await (const line of ndjsonLines(res.body)) {
    try {
      const record = JSON.parse(line);
      processed += 1;
      if (processed % 1000 === 0) console.log('processed', processed);
    } catch {
      failed += 1; // 单行损坏不影响整条流
    }
  }
  return { processed, failed };
}

const stats = await ingest('https://api.example.com/events/stream');
console.log('ingest done:', stats);