fix: buffer progress chunks so a split event terminator doesn't merge events

[emaglic-xa-eg]

Fixes #38.

Problem

When a network chunk boundary lands inside an event's \n\n terminator, two events get concatenated into one. The merged payload is invalid for consumers that parse data (for example JSON.parse), and when the trailing event is the stream's final one, the preceding event is dropped with no error at all.

Cause

onStreamProgress split each progress slice independently:

data = data.substring(this.progress);
this.progress += data.length;
data.split(/(\r\n|\r|\n){2}/g).forEach((part) => this.parseEventData(part, observer));

A progress slice can end partway through a terminator, so neither slice contains the full \n\n. The boundary is never recognized, the first newline is buffered as event content, and the next event is appended before the flush happens, so both events land in one chunk and merge.

Fix

Buffer the raw text across progress slices and only emit once a full terminator is present. Anything after the last complete terminator stays buffered for the next slice, so a split terminator just means "keep buffering" instead of "merge":

const fresh = data.substring(this.progress);
this.progress += fresh.length;
this.chunk += fresh;

let match: RegExpExecArray | null;
const boundary = /\r\n\r\n|\n\n|\r\r/;
while ((match = boundary.exec(this.chunk))) {
  const event = this.chunk.slice(0, match.index);
  this.chunk = this.chunk.slice(match.index + match[0].length);
  this.dispatchStreamData(this.parseEventChunk(event), observer);
}

I also reset the buffer on the Sent event. A retryWhen reconnection skips onStreamCompleted, which is the only place the buffer was cleared, so without this a partial event from a dropped connection would merge into the first event of the reconnected stream. Sent fires on every (re)connection, so it is a reliable per-connection reset point.

The old parseEventData helper was the only caller of the split path and is now unused, so I removed it.

Repro

Dependency-free, runs with node repro.mjs. It uses the library's real parseEventChunk / parseChunkLine and only swaps onStreamProgress (current vs this PR), fed the same cumulative partialText snapshots the browser delivers, with a boundary inside the terminator:

repro.mjs

const enc = (v) => `data: ${JSON.stringify(v)}`;
const EVENT1 = enc({ type: 'message', value: 'first' });
const EVENT2 = enc({ type: 'message', value: 'second' });
const STREAM = `${EVENT1}\n\n${EVENT2}\n\n`;

// Cumulative partialText snapshots. Delivery 1 ends one byte into the first
// terminator, so the network boundary splits "\n\n".
const SPLIT = EVENT1.length + 1;
const DELIVERIES = [STREAM.slice(0, SPLIT), STREAM];

const SEPARATOR = ':';

function parseChunkLine(line, event) {
  const index = line.indexOf(SEPARATOR);
  if (index <= 0) return;
  const field = line.substring(0, index);
  if (Object.keys(event).findIndex((key) => key === field) === -1) return;
  let data = line.substring(index + 1).replace(/^\s/, '');
  if (field === 'data') data = event.data + data;
  event[field] = data;
}

function parseEventChunk(chunk) {
  if (!chunk || chunk.length === 0) return;
  const chunkEvent = { id: undefined, data: '', event: 'message' };
  chunk.split(/\n|\r\n|\r/).forEach((line) => parseChunkLine(line.trim(), chunkEvent));
  return { type: chunkEvent.event, data: chunkEvent.data };
}

// current onStreamProgress + its parseEventData helper
function runCurrent(deliveries) {
  const emitted = [];
  const state = { progress: 0, chunk: '' };
  const dispatch = (e) => { if (e && e.data) emitted.push(e.data); };
  const parseEventData = (part) => {
    if (part.trim().length === 0) { dispatch(parseEventChunk(state.chunk)); state.chunk = ''; }
    else { state.chunk += part; }
  };
  for (const data of deliveries) {
    if (!data) continue;
    const slice = data.substring(state.progress);
    state.progress += slice.length;
    slice.split(/(\r\n|\r|\n){2}/g).forEach((part) => parseEventData(part));
  }
  return emitted;
}

// fixed onStreamProgress
function runFixed(deliveries) {
  const emitted = [];
  const state = { progress: 0, chunk: '' };
  const dispatch = (e) => { if (e && e.data) emitted.push(e.data); };
  for (const data of deliveries) {
    if (!data) continue;
    const fresh = data.substring(state.progress);
    state.progress += fresh.length;
    state.chunk += fresh;
    let match;
    const boundary = /\r\n\r\n|\n\n|\r\r/;
    while ((match = boundary.exec(state.chunk))) {
      const event = state.chunk.slice(0, match.index);
      state.chunk = state.chunk.slice(match.index + match[0].length);
      dispatch(parseEventChunk(event));
    }
  }
  return emitted;
}

function report(label, emitted) {
  console.log(`\n${label}: ${emitted.length} event(s) emitted`);
  emitted.forEach((data, i) => {
    let ok = true;
    try { JSON.parse(data); } catch { ok = false; }
    console.log(`  [${i}] JSON.parse ${ok ? 'OK  ' : 'FAIL'}  data=${data}`);
  });
}

report('CURRENT', runCurrent(DELIVERIES));
report('FIXED  ', runFixed(DELIVERIES));

Output:

CURRENT: 1 event(s) emitted
  [0] JSON.parse FAIL  data={"type":"message","value":"first"}{"type":"message","value":"second"}

FIXED  : 2 event(s) emitted
  [0] JSON.parse OK    data={"type":"message","value":"first"}
  [1] JSON.parse OK    data={"type":"message","value":"second"}

Validation

• npm run build:prod passes.
• Beyond the single boundary above, I verified the fix offline across every boundary position in a two-slice split and across fixed-size chunkings (from byte-by-byte up through large buffers): the current code merges at the terminator-split positions, the patched code stays clean at all of them.

Tests

The library doesn't have a test harness wired up right now (no specs, and karma/jasmine aren't in the dev dependencies), so I kept this PR to the fix plus the standalone repro above rather than adding a test framework unprompted. Happy to follow up with a spec, or to set up Karma/Jasmine in a separate PR, if you'd like that.