Add support for AsyncIterable as input #46

huntharo · 2021-08-25T20:23:12Z

Overview

Expand capabilities to handle both async and sync iterables - Fixes Support Async Iterable Protocol #20

Open Issues

Duplicating the tests for the async versions doesn't feel right - Let me know if there is a better way to handle this with ava
There is a problem in the original functionality around the stop-on-error test that needs to be resolved
- The test has a bug in that neither p-map nor the test mapper function, which is provided for that specific test and is not the common version, await the calling of the delay() function for item 2
- This means that item 2 will only ever return a Promise, never a value
- When this is fixed to await the function call, it causes the returned results to change from [1, 3] to [1, 3, 2]
- Fixing this test makes it clear that the concurrency setup loop does not wait for mappers, so the only way to make this test fail would be to have enough items that eventually the initial iteration could not reach the last items in the test array before the 100 ms delay on the exception expires
- With concurrency = 1 this test behaves as expected since there are not unlimited unawaited promises created
- Let me know your thoughts on how to resolve this as it becomes more apparent with the behavior changes needed for asyncIterable (where the concurrency setup iteration of asyncIterable has to be awaited to prevent inifinte runners from being created)
- The test, in the state it's in in this PR, does not actually demonstrate that stop on error works as intended

sindresorhus · 2021-09-07T10:31:30Z

Duplicating the tests for the async versions doesn't feel right

That's how it's done. There's good way of mixing async and sync in JS.

sindresorhus · 2021-09-07T10:32:41Z

There is a problem in the original functionality around the stop-on-error test that needs to be resolved

Which test exactly? Can you link to it?

index.js

huntharo · 2021-09-07T16:25:05Z

There is a problem in the original functionality around the stop-on-error test that needs to be resolved

Which test exactly? Can you link to it?

This test: https://github.com/sindresorhus/p-map/blob/main/test.js#L121-L135

Stop on error is problematic... there can be many concurrent mappers running and they will continue to run even if iteration is "stopped" after one of them throws. If the continuing mappers also throw they will have nowhere to throw to that can be caught. So it seems that the best way to handle an exception in one mapper is to check if any other mappers are running, push the exception onto an array, then let the other mappers either succeed or fail, and the last of them throws and AggregateError with the exceptions from all of them or just throws the single error if only one of them failed. It definitely should not throw until all of the mappers have finished though as that could lead to the consumer restarting the pMap while mappers from the prior run are still active.

Ultimately what I ended up thinking was: if you need clean error handling then you should never allow the mapper to throw and instead handle the error yourself in the mapper. However, I don't think there is a way for the mapper to signal that iteration should stop early.

Current form of the test in the PR

The changes here are:
- Add an async closure for the delay call so it can be detected as a function to be awaited - Checking instanceof Promise or ClearablePromise didn't work in this case
- When the input value is a function, await the call of the function
- Change the expected return value to 1, 3, 2 - Which is "wrong" as it shows that all values up to concurrency get mapped even if one of them throws an error before the others finish
The test does not demonstrate that iteration / mapper invocation stops after a mapper throws because the default concurrency is infinite and all of the mappers will be invoked before any one of them throws an exception - We can change this to have concurrency 1 and it will then return a result of [1], but all 3 mappers will still be invoked and will run to completion.

test('do not run mapping after stop-on-error happened', async t => {
	const input = [1, async () => delay(300, {value: 2}), 3];
	const mappedValues = [];
	await t.throwsAsync(
		pMap(input, async value => {
			value = typeof value === 'function' ? await value() : value;
			mappedValues.push(value);
			if (value === 1) {
				await delay(100);
				throw new Error('Oops!');
			}
		})
	);
	await delay(500);
	t.deepEqual(mappedValues, [1, 3, 2]);
});

sindresorhus · 2021-09-07T16:42:54Z

This test: https://github.com/sindresorhus/p-map/blob/main/test.js#L121-L135

// @papb ^

huntharo · 2021-09-07T19:39:05Z

This test: https://github.com/sindresorhus/p-map/blob/main/test.js#L121-L135

// @papb ^

I added a PR with a failing test case that shows that multiple concurrent mapper exceptions result in only one of those errors being catchable:

#49

I think the stopOnError true/false logic can be somewhat consolidated so that both reject with an AggregateError when mappers have all stopped, but stopOnError true immediately stops invoking more mappers and iterating more source items (if applicable, as in concurrency !== Infinity).

sindresorhus · 2021-09-09T17:36:38Z

I think the stopOnError true/false logic can be somewhat consolidated so that both reject with an AggregateError when mappers have all stopped

Per #49 (comment), stopOnError: true should only return a single error. This is how Promise.all works too.

- Expand capabilities to handle both async and sync iterables - There is a problem in the original functionality around the `stop-on-error` test that needs to be resolved - The test has a bug in that neither p-map nor the test mapper function, which is provided for that specific test and is not the common version, await the calling of the delay() function for item 2 - This means that item 2 will only ever return a Promise, never a value - When this is fixed to await the function call, it causes the returned results to change from `[1, 3]` to `[1, 3, 2]` - Fixing this test makes it clear that the `concurrency setup` loop does not wait for mappers, so the only way to make this test fail would be to have enough items that eventually the initial iteration could not reach the last items in the test array before the 100 ms delay on the exception expires - With `concurrency = 1` this test behaves as expected since there are not unlimited unawaited promises created - Let me know your thoughts on how to resolve this as it becomes more apparent with the behavior changes needed for asyncIterable (where the concurrency setup iteration of asyncIterable has to be awaited to prevent inifinte runners from being created) - The test, in the state it's in in this PR, does not actually demonstrate that stop on error works as intended

Co-authored-by: Sindre Sorhus <sindresorhus@gmail.com>

huntharo · 2021-09-12T20:11:34Z

I believe all outstanding issues have been addressed now.

index.js

sindresorhus · 2021-09-13T05:47:32Z

Can you update the readme too?

- Also distinguish from items returning a promise - Update the TS types to indicate that iterated items can actually be a `Promise`

huntharo · 2021-09-16T18:03:39Z

Can you update the readme too?

The readme has been updated.

sindresorhus · 2021-09-23T17:49:41Z

index.js

 		let isIterableDone = false;
 		let resolvingCount = 0;
 		let currentIndex = 0;
+		const iterator = iterable[Symbol.iterator] === undefined ? iterable[Symbol.asyncIterator]() : iterable[Symbol.iterator]();


I think we should throw a TypeError with a human-friendly error if the input is neither an iterable or async iterable.

What do you think?

Inner function within the function so iterator can remain const? File scope function?

Nested ? operator?

if/else if/else with throw and iterator being non-const?

Also... when you say "TypeError with a human-friendly error"... did you mean a human friendly error-message? Or some specific field you want set on the error?

done... i added a check for the iterator type at the top, similar to mapper / concurrency checks, which means that the existing code cannot have an error as it would have already thrown out

sindresorhus · 2021-09-23T17:50:28Z

index.js

-				break;
-			}
+		// Create the concurrent runners in a detached (non-awaited)
+		// promise.  We need this so we can await the next() calls


Suggested change

// promise. We need this so we can await the next() calls

// promise. We need this so we can await the `next()` calls

sindresorhus · 2021-09-23T17:50:58Z

index.js

+		// to stop creating runners before hitting the concurrency limit
+		// if the iterable has already been marked as done.
+		// NOTE: We *must* do this for async iterators otherwise we'll spin up
+		// infinite next() calls by default and never start the event loop.


Suggested change

// infinite next() calls by default and never start the event loop.

// infinite `next()` calls by default and never start the event loop.

done (plus more cases)

sindresorhus · 2021-09-23T17:52:00Z

readme.md

-Iterated over concurrently in the `mapper` function.
+Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element.  Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item.
+
+Asynchoronous iterables (distinguishing from iterables that return `Promise` that resolves to an item) can be used when the next item may not be ready without waiting for an asynchronous process to complete and/or the end of the iterable may be reached after the asynchornous proces completes (e.g. reading from a remote queue when the queue has reached empty, or reading lines from a stream).



You also need to update index.d.ts

sindresorhus · 2021-09-23T17:52:34Z

readme.md


-Iterated over concurrently in the `mapper` function.
+Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element.  Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item.


Suggested change

Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element. Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item.

Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element. Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item.

sindresorhus · 2021-09-23T17:54:29Z

readme.md

-Iterated over concurrently in the `mapper` function.
+Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element.  Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item.
+
+Asynchoronous iterables (distinguishing from iterables that return `Promise` that resolves to an item) can be used when the next item may not be ready without waiting for an asynchronous process to complete and/or the end of the iterable may be reached after the asynchornous proces completes (e.g. reading from a remote queue when the queue has reached empty, or reading lines from a stream).


Suggested change

Asynchoronous iterables (distinguishing from iterables that return `Promise` that resolves to an item) can be used when the next item may not be ready without waiting for an asynchronous process to complete and/or the end of the iterable may be reached after the asynchornous proces completes (e.g. reading from a remote queue when the queue has reached empty, or reading lines from a stream).

Asynchronous iterables (different from synchronous iterables that return `Promise` that resolves to an item) can be used when the next item may not be ready without waiting for an asynchronous process to complete and/or the end of the iterable may be reached after the asynchronous process completes. For example, reading from a remote queue when the queue has reached empty, or reading lines from a stream.

sindresorhus · 2021-09-23T17:55:14Z

test.js

+	[async () => {
+		return 10;
+	}, 300],


Suggested change

[async () => {

return 10;

}, 300],

[async () => 10, 300],

sindresorhus · 2021-09-23T17:55:43Z

test.js

+
+	async * [Symbol.asyncIterator]() {
+		for (let i = 0; i < this.data.length; i++) {
+			// Add a delay between each item iterated


Suggested change

// Add a delay between each item iterated

// Add a delay between each iterated item

sindresorhus · 2021-09-23T17:55:57Z

test.js

+	}
+
+	async * [Symbol.asyncIterator]() {
+		for (let i = 0; i < this.data.length; i++) {


Suggested change

for (let i = 0; i < this.data.length; i++) {

for (let index = 0; index < this.data.length; index++) {

sindresorhus · 2021-09-24T02:22:23Z

Looks good. Thanks for contributing this :)

huntharo · 2021-09-24T13:04:16Z

Looks good. Thanks for contributing this :)

No problem. Thanks for this (and other) modules and for the timely reviews and merges!

sindresorhus requested changes Sep 7, 2021

View reviewed changes

index.js Outdated Show resolved Hide resolved

index.js Outdated Show resolved Hide resolved

index.js Outdated Show resolved Hide resolved

sindresorhus changed the title ~~Implement #20 - Async Iterable Support~~ Add support for async iterables as input Sep 7, 2021

huntharo and others added 3 commits September 12, 2021 15:46

Update index.js

8477a7a

Co-authored-by: Sindre Sorhus <sindresorhus@gmail.com>

Prevent skip removal logic from being run twice

d5a0b26

huntharo force-pushed the fix/20 branch from ef6558f to d5a0b26 Compare September 12, 2021 19:50

sindresorhus reviewed Sep 13, 2021

View reviewed changes

index.js Outdated Show resolved Hide resolved

huntharo added 2 commits September 16, 2021 13:42

Rename per request

6d4156d

Update readme to document async iterables

8b6e6f0

- Also distinguish from items returning a promise - Update the TS types to indicate that iterated items can actually be a `Promise`

huntharo changed the title ~~Add support for async iterables as input~~ Add support for AsyncIterable as input Sep 17, 2021

sindresorhus requested changes Sep 23, 2021

View reviewed changes

Attempt to address all requests

c9e7112

sindresorhus merged commit a24e909 into sindresorhus:main Sep 24, 2021

huntharo deleted the fix/20 branch September 24, 2021 13:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for AsyncIterable as input #46

Add support for AsyncIterable as input #46

huntharo commented Aug 25, 2021 •

edited

sindresorhus commented Sep 7, 2021

sindresorhus commented Sep 7, 2021

huntharo commented Sep 7, 2021

sindresorhus commented Sep 7, 2021

huntharo commented Sep 7, 2021

sindresorhus commented Sep 9, 2021

huntharo commented Sep 12, 2021

sindresorhus commented Sep 13, 2021

huntharo commented Sep 16, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus Sep 23, 2021

huntharo Sep 23, 2021

sindresorhus commented Sep 24, 2021

huntharo commented Sep 24, 2021

	// promise. We need this so we can await the next() calls
	// promise. We need this so we can await the `next()` calls

	// infinite next() calls by default and never start the event loop.
	// infinite `next()` calls by default and never start the event loop.


		Iterated over concurrently in the `mapper` function.
		Synchronous or asynchronous iterable that is iterated over concurrently, calling the `mapper` function for each element. Each iterated item is `await`'d before the `mapper` is invoked so the iterable may return a `Promise` that resolves to an item.

	// Add a delay between each item iterated
	// Add a delay between each iterated item

	for (let i = 0; i < this.data.length; i++) {
	for (let index = 0; index < this.data.length; index++) {

Add support for AsyncIterable as input #46

Add support for AsyncIterable as input #46

Conversation

huntharo commented Aug 25, 2021 • edited

Overview

Open Issues

sindresorhus commented Sep 7, 2021

sindresorhus commented Sep 7, 2021

huntharo commented Sep 7, 2021

Current form of the test in the PR

sindresorhus commented Sep 7, 2021

huntharo commented Sep 7, 2021

sindresorhus commented Sep 9, 2021

huntharo commented Sep 12, 2021

sindresorhus commented Sep 13, 2021

huntharo commented Sep 16, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sindresorhus commented Sep 24, 2021

huntharo commented Sep 24, 2021

huntharo commented Aug 25, 2021 •

edited