Add Async Iterable support

[Tigge]

No description provided.

[kitten]

Hiya 👋

I'm not quite sure why this was opened without a description, and for now, while I'm open to reopening this of course if things change, I'll close this PR for now.

This basically has two reasons.

The first is, that I skipped support of adding a toAsyncIterable primitive on purpose. It's just not something I see as useful right now, and I haven't seen a lot of cases where it'd be useful, in my opinion, just yet. I don't exclude there being a possibility of this in the future. However, any implementation would have to be better at backpressure, meaning, it'd have to pull only when a value has resolved, and not eagerly. Currently, your implementation for instance does not handle the case of only pulling when a next value is requested.

However, again, I don't think this is something I'd want to add to the library right now.

As per fromAsyncIterable, Wonka already contains this, together with support for non-async-iterables:

wonka/src/sources.ts

Lines 9 to 88 in 34ee8d7

	export function fromAsyncIterable<T>(iterable: AsyncIterable<T>): Source<T> {
	return sink => {
	const iterator = iterable[Symbol.asyncIterator]();
	let ended = false;
	let looping = false;
	let pulled = false;
	let next: IteratorResult<T>;
	sink(
	start(async signal => {
	if (signal === TalkbackKind.Close) {
	ended = true;
	if (iterator.return) iterator.return();
	} else if (looping) {
	pulled = true;
	} else {
	for (pulled = looping = true; pulled && !ended; ) {
	if ((next = await iterator.next()).done) {
	ended = true;
	if (iterator.return) await iterator.return();
	sink(SignalKind.End);
	} else {
	try {
	pulled = false;
	sink(push(next.value));
	} catch (error) {
	if (iterator.throw) {
	if ((ended = !!(await iterator.throw(error)).done)) sink(SignalKind.End);
	} else {
	throw error;
	}
	}
	}
	}
	looping = false;
	}
	})
	);
	};
	}
	export function fromIterable<T>(iterable: Iterable<T> | AsyncIterable<T>): Source<T> {
	if (iterable[Symbol.asyncIterator]) return fromAsyncIterable(iterable as AsyncIterable<T>);
	return sink => {
	const iterator = iterable[Symbol.iterator]();
	let ended = false;
	let looping = false;
	let pulled = false;
	let next: IteratorResult<T>;
	sink(
	start(signal => {
	if (signal === TalkbackKind.Close) {
	ended = true;
	if (iterator.return) iterator.return();
	} else if (looping) {
	pulled = true;
	} else {
	for (pulled = looping = true; pulled && !ended; ) {
	if ((next = iterator.next()).done) {
	ended = true;
	if (iterator.return) iterator.return();
	sink(SignalKind.End);
	} else {
	try {
	pulled = false;
	sink(push(next.value));
	} catch (error) {
	if (iterator.throw) {
	if ((ended = !!iterator.throw(error).done)) sink(SignalKind.End);
	} else {
	throw error;
	}
	}
	}
	}
	looping = false;
	}
	})
	);
	};
	}

As you see, it also has a bit of a different call signature and doesn't rely on catch(console.error), instead escalating this to be an uncaught error instead. Details like these are a little more important to prevent us handling errors where they shouldn't happen, where catch actually tells the runtime that the error is in fact being handled.

Another small detail, which for-await-of actually doesn't quite handle that well is the behaviour of iterator.return and iterator.throw, which is in there because it can be used in case of cancellation, and in case of skipping and swallowing errors down the chain.

[Tigge]

Hi, sorry for not providing a description, will do better next time. I didn't notice that there was already a fromAsyncIterable, we will start using that instead of our own version. It seem to be missing from the documentation though.

I can shed some light on our use case. We are using this together with graphql.js for subscriptions which are built around AsyncIterable - so hence our need for something like this for turning wonka streams into something that can be consumed by graphql.js. Since we are already using urql and wonka it seemed like a good fit to use it there as well.

We recently upgraded to version 6 as well which was a nice change with the typescript codebase - at least for me. Thank you!