Function iterSSEChunks

packages/baseai/src/dev/utils/stream/stream.ts:246–278

* Given an async iterable iterator, iterates over it and yields full * SSE chunks, i.e. yields when a double new-line is encountered.

(
	iterator: AsyncIterableIterator<Bytes>
)

 * SSE chunks, i.e. yields when a double new-line is encountered.
 */
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;
	}
}

function findDoubleNewlineIndex(buffer: Uint8Array): number {
	// This function searches the buffer for the end patterns (\r\r, \n\n, \r\n\r\n)