Function iterSSEChunks

packages/core/src/common/stream.ts:251–281

* Asynchronously iterates over SSE (Server-Sent Events) chunks and yields the data as Uint8Array. * * Given an async iterable iterator, iterates over it and yields full * SSE chunks, i.e. yields when a double new-line is encountered. * * @param iterator - The async iterator that provides the SS

(iterator: AsyncIterableIterator<Bytes>)

 * @returns An async generator that yields Uint8Array chunks.
 */
async function* iterSSEChunks(iterator: AsyncIterableIterator<Bytes>): AsyncGenerator<Uint8Array> {
	let data = new Uint8Array();

	for await (const chunk of iterator) {
		if (chunk == null) {
			continue;
		}

		const binaryChunk =
			chunk instanceof ArrayBuffer
				? new Uint8Array(chunk)
				: typeof chunk === 'string'
					? new TextEncoder().encode(chunk)
					: chunk;

		let newData = new Uint8Array(data.length + binaryChunk.length);
		newData.set(data);
		newData.set(binaryChunk, data.length);
		data = newData;

		let patternIndex;
		while ((patternIndex = findDoubleNewlineIndex(data)) !== -1) {
			yield data.slice(0, patternIndex);
			data = data.slice(patternIndex);
		}
	}

	if (data.length > 0) {
		yield data;
	}
}

/**
 * Finds the index of the first occurrence of a double newline pattern (\r\r, \n\n, \r\n\r\n) in the given buffer.