
type Observer<T> = web3n.Observer<T>;

export function streamFrom<T>(
	startWatching: (obs: Observer<T>) => (() => void)
): ReadableStream<T> {
	let unsub: () => void;
	return new ReadableStream<T>({
		start(ctrl) {
			unsub = startWatching({
				next: v => ctrl.enqueue(v),
				complete: () => ctrl.close(),
				error: err => ctrl.error(err)
			});
		},
		cancel: () => unsub()
	});
}

export function map<I, O>(fn: (v: I) => O): TransformStream<I, O> {
	return new TransformStream<I, O>({
		transform: (v, ctrl) => ctrl.enqueue(fn(v))
	});
}

export function filter<T>(predicate: (v: T) => boolean): TransformStream<T, T> {
	return new TransformStream<T, T>({
		transform: (v, ctrl) => {
			if (predicate(v)) {
				ctrl.enqueue(v);
			}
		}
	});
}

export function doWithEach<T>(fn: (v: T) => void): TransformStream<T, T> {
	return new TransformStream<T, T>({
		transform: (v, ctrl) => {
			try {
				fn(v);
			} catch (err) {
				console.error(err);
			}
			ctrl.enqueue(v);
		}
	});
}

export function doAndAwaitWithEach<T>(
	fn: (v: T) => Promise<void>
): TransformStream<T, T> {
	return new TransformStream<T, T>({
		transform: async (v, ctrl) => {
			try {
				await fn(v);
			} catch (err) {
				console.error(err);
			}
			ctrl.enqueue(v);
		}
	});
}

export async function* readerToIterable<T>(
	reader: ReadableStreamDefaultReader<T>
) {
	while (true) {
		const { done, value } = await reader.read();
		if (done) {
			break;
		} else {
			yield value;
		}
	}
}

