Method fromReadableStream

packages/core/src/common/stream.ts:101–142

* Generates a Stream from a newline-separated ReadableStream * where each item is a JSON value. * * @template Item - The type of items in the stream. * @param {ReadableStream} readableStream - The readable stream to create the stream from. * @param {AbortController} controller - The abort

(readableStream: ReadableStream, controller: AbortController)

	 * @returns {Stream<Item>} - The created stream.
	 */
	static fromReadableStream<Item>(readableStream: ReadableStream, controller: AbortController) {
		let consumed = false;

		async function* iterLines(): AsyncGenerator<string, void, unknown> {
			const lineDecoder = new LineDecoder();

			const iter = readableStreamAsyncIterable<Bytes>(readableStream);
			for await (const chunk of iter) {
				for (const line of lineDecoder.decode(chunk)) {
					yield line;
				}
			}

			for (const line of lineDecoder.flush()) {
				yield line;
			}
		}

		async function* iterator(): AsyncIterator<Item, any, undefined> {
			if (consumed) {
				throw new Error('Cannot iterate over a consumed stream, use `.tee()` to split the stream.');
			}
			consumed = true;
			let done = false;
			try {
				for await (const line of iterLines()) {
					if (done) continue;
					if (line) yield JSON.parse(line);
				}
				done = true;
			} catch (e) {
				// If the user calls `stream.controller.abort()`, we should exit without throwing.
				if (e instanceof Error && e.name === 'AbortError') return;
				throw e;
			} finally {
				// If the user `break`s, abort the ongoing request.
				if (!done) controller.abort();
			}
		}

		return new Stream(iterator, controller);
	}

	[Symbol.asyncIterator](): AsyncIterator<Item> {
		return this.iterator();