Add pFilterIterable (#7)

index.d.ts

@@ -1,4 +1,4 @@
-import {Options} from 'p-map';
+import type {Options} from 'p-map';
 
 /**
 Filter promises concurrently.
@@ -38,4 +38,48 @@ export default function pFilter<ValueType>(
 	options?: Options
 ): Promise<ValueType[]>;
 
+/**
+Filter promises concurrently.
+
+@param input - Iterated over concurrently in the `filterer` function.
+@param filterer - The filterer function that decides whether an element should be included into result.
+@param options - See the [`p-map` options](https://github.com/sindresorhus/p-map#options).
+@returns An async iterable that iterates over the promises in `iterable` and ones returned from `filterer` concurrently, calling `filterer` for each element.
+
+@example
+```
+import {pFilterIterable} from 'p-filter';
+import getWeather from 'get-weather'; // Not a real module
+
+async function * getPlaces() {
+	const name = await getCapital('Norway');
+
+	yield name;
+	yield 'Bangkok, Thailand';
+	yield 'Berlin, Germany';
+	yield 'Tokyo, Japan';
+}
+
+const places = getPlaces();
+
+const filterer = async place => {
+	const weather = await getWeather(place);
+	return weather.temperature > 30;
+};
+
+const result = await pFilterIterable(places, filterer);
+
+console.log(result);
+//=> ['Bangkok, Thailand']
+```
+*/
+export function pFilterIterable<ValueType>(
+	input: Iterable<ValueType | PromiseLike<ValueType>>,
+	filterer: (
+		element: ValueType,
+		index: number
+	) => boolean | PromiseLike<boolean>,
+	options?: Options
+): AsyncIterable<ValueType>;
+
 export {Options} from 'p-map';

index.js

@@ -1,4 +1,4 @@
-import pMap from 'p-map';
+import pMap, {pMapIterable} from 'p-map';
 
 export default async function pFilter(iterable, filterer, options) {
 	const values = await pMap(
@@ -9,3 +9,21 @@ export default async function pFilter(iterable, filterer, options) {
 
 	return values.filter(value => Boolean(value[0])).map(value => value[1]);
 }
+
+export function pFilterIterable(iterable, filterer, options) {
+	const values = pMapIterable(
+		iterable,
+		(element, index) => Promise.all([filterer(element, index), element]),
+		options,
+	);
+
+	return {
+		async * [Symbol.asyncIterator]() {
+			for await (const [value, element] of values) {
+				if (value) {
+					yield element;
+				}
+			}
+		},
+	};
+}

index.test-d.ts

@@ -1,5 +1,5 @@
 import {expectType} from 'tsd';
-import pFilter from './index.js';
+import pFilter, {pFilterIterable} from './index.js';
 
 const places = [
 	'Bangkok, Thailand',
@@ -15,3 +15,13 @@ expectType<Promise<string[]>>(
 expectType<Promise<number[]>>(
 	pFilter(new Set([1, 2]), number => number > 1, {concurrency: 1}),
 );
+
+expectType<AsyncIterable<string>>(
+	pFilterIterable(places, async place =>
+		place === 'Bangkok, Thailand' ? true : Promise.resolve(false),
+	),
+);
+
+expectType<AsyncIterable<number>>(
+	pFilterIterable(new Set([1, 2]), number => number > 1, {concurrency: 1}),
+);

package.json

@@ -38,11 +38,11 @@
 		"bluebird"
 	],
 	"dependencies": {
-		"p-map": "^5.1.0"
+		"p-map": "^7.0.0"
 	},
 	"devDependencies": {
-		"ava": "^3.15.0",
-		"tsd": "^0.17.0",
-		"xo": "^0.44.0"
+		"ava": "^6.0.1",
+		"tsd": "^0.30.0",
+		"xo": "^0.56.0"
 	}
 }

readme.md

@@ -56,6 +56,65 @@ The filterer function that decides whether an element should be included into re
 
 Type: `object`
 
+See the [`p-map` options](https://github.com/sindresorhus/p-map#options).
+
+##### concurrency
+
+Type: `number`\
+Default: `Infinity`\
+Minimum: `1`
+
+The number of concurrently pending promises returned by `filterer`.
+
+### pFilterIterable(iterable, filterer, options?)
+
+Returns an async iterable that iterates over the promises in `iterable` and ones returned from `filterer` concurrently, calling `filterer` for each element.
+
+#### Usage
+```js
+import {pFilterIterable} from 'p-filter';
+import getWeather from 'get-weather'; // Not a real module
+
+async function * getPlaces() {
+	const name = await getCapital('Norway');
+
+	yield name;
+	yield 'Bangkok, Thailand';
+	yield 'Berlin, Germany';
+	yield 'Tokyo, Japan';
+}
+
+const places = getPlaces();
+
+const filterer = async place => {
+	const weather = await getWeather(place);
+	return weather.temperature > 30;
+};
+
+const result = await pFilterIterable(places, filterer);
+
+console.log(result);
+//=> ['Bangkok, Thailand']
+```
+
+#### iterable
+
+Type: `Iterable<Promise|any>`
+
+Iterated over concurrently in the `filterer` function.
+
+#### filterer(element, index)
+
+Type: `Function`
+
+The filterer function that decides whether an element should be included into result. Expected to return `boolean | Promise<boolean>`.
+
+#### options
+
+Type: `object`
+
+See the [`p-map` options](https://github.com/sindresorhus/p-map#options).
+
 ##### concurrency
 
 Type: `number`\

test.js

@@ -1,12 +1,46 @@
 import test from 'ava';
-import pFilter from './index.js';
+import pFilter, {pFilterIterable} from './index.js';
 
 // See `p-map` for more comprehensive tests
 test('main', async t => {
-	t.deepEqual(await pFilter([Promise.resolve(1), 2, 3, 4], x => x % 2), [1, 3]);
-	t.deepEqual(await pFilter([1, 2, 3, 4], x => Promise.resolve(x % 2)), [1, 3]);
+	t.deepEqual(
+		await pFilter([Promise.resolve(1), 2, 3, 4], x => x % 2),
+		[1, 3],
+	);
+	t.deepEqual(
+		await pFilter([1, 2, 3, 4], x => Promise.resolve(x % 2)),
+		[1, 3],
+	);
 });
 
 test('handles empty iterable', async t => {
 	t.deepEqual(await pFilter([]), []);
 });
+
+test('pFilterIterable', async t => {
+	const rangeIterable = {
+		async * [Symbol.asyncIterator]() {
+			yield 1;
+			yield 2;
+			yield 3;
+			yield 4;
+		},
+	};
+	const iterable = pFilterIterable(rangeIterable, x => x % 2);
+	const results = [];
+	for await (const x of iterable) {
+		results.push(x);
+	}
+
+	t.deepEqual(results, [1, 3]);
+
+	const iterable2 = pFilterIterable(rangeIterable, x =>
+		Promise.resolve(x % 2),
+	);
+	const results2 = [];
+	for await (const x of iterable2) {
+		results2.push(x);
+	}
+
+	t.deepEqual(results2, [1, 3]);
+});