-
Notifications
You must be signed in to change notification settings - Fork 72
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix(flatMap): default flatMap concurrent parameter to 1 (#346)
* fix(flatMap): default flatMap concurrent parameter to 1 * feat(switchall): add switchAll operator * fix(concatmap): align concatMap source iteration behavior with concatAll
- Loading branch information
Showing
23 changed files
with
219 additions
and
45 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| import { hasNext, noNext } from '../asynciterablehelpers'; | ||
| import { of, range, sequenceEqual, throwError } from 'ix/asynciterable'; | ||
| import { map, tap, concatMap } from 'ix/asynciterable/operators'; | ||
|
|
||
| test('AsyncIterable#concatMap with range', async () => { | ||
| const xs = of(1, 2, 3); | ||
| const ys = xs.pipe(concatMap(async (x) => range(0, x))); | ||
|
|
||
| const it = ys[Symbol.asyncIterator](); | ||
| hasNext(it, 0); | ||
| hasNext(it, 0); | ||
| hasNext(it, 1); | ||
| hasNext(it, 0); | ||
| hasNext(it, 1); | ||
| hasNext(it, 2); | ||
| noNext(it); | ||
| }); | ||
|
|
||
| test('AsyncIterable#concatMap order of effects', async () => { | ||
| let i = 0; | ||
| const res = range(0, 3).pipe( | ||
| tap({ next: async () => ++i }), | ||
| concatMap((x) => range(0, x + 1)), | ||
| map((x) => i + ' - ' + x) | ||
| ); | ||
|
|
||
| expect( | ||
| await sequenceEqual(res, of('1 - 0', '2 - 0', '2 - 1', '3 - 0', '3 - 1', '3 - 2')) | ||
| ).toBeTruthy(); | ||
| }); | ||
|
|
||
| test('AsyncIterable#concatMap selector returns throw', async () => { | ||
| const err = new Error(); | ||
| const xs = of(1, 2, 3); | ||
| const ys = xs.pipe(concatMap(async (x) => (x < 3 ? range(0, x) : throwError(err)))); | ||
|
|
||
| const it = ys[Symbol.asyncIterator](); | ||
| hasNext(it, 0); | ||
| hasNext(it, 0); | ||
| hasNext(it, 1); | ||
| await expect(it.next()).rejects.toThrow(err); | ||
| }); | ||
|
|
||
| test('AsyncIterable#concatMap with error throws', async () => { | ||
| const err = new Error(); | ||
| const xs = throwError(err); | ||
| const ys = xs.pipe(concatMap((x) => range(0, x))); | ||
|
|
||
| const it = ys[Symbol.asyncIterator](); | ||
| await expect(it.next()).rejects.toThrow(err); | ||
| }); | ||
|
|
||
| test('AsyncIterable#concatMap selector throws error', async () => { | ||
| const err = new Error(); | ||
| const xs = of(1, 2, 3); | ||
| const ys = xs.pipe( | ||
| concatMap(async (x) => { | ||
| if (x < 3) { | ||
| return range(0, x); | ||
| } | ||
| throw err; | ||
| }) | ||
| ); | ||
|
|
||
| const it = ys[Symbol.asyncIterator](); | ||
| hasNext(it, 0); | ||
| hasNext(it, 0); | ||
| hasNext(it, 1); | ||
| await expect(it.next()).rejects.toThrow(err); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,10 +1,10 @@ | ||
| import '../asynciterablehelpers'; | ||
| import { of, empty, isEmpty } from 'ix/asynciterable'; | ||
|
|
||
| test('Iterable#isEmpty empty', async () => { | ||
| test('AsyncIterable#isEmpty empty', async () => { | ||
| expect(await isEmpty(empty())).toBeTruthy(); | ||
| }); | ||
|
|
||
| test('Iterable#isEmpty not-empty', async () => { | ||
| test('AsyncIterable#isEmpty not-empty', async () => { | ||
| expect(await isEmpty(of(1))).toBeFalsy(); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| import '../asynciterablehelpers'; | ||
| import { of, toArray } from 'ix/asynciterable'; | ||
| import { switchAll, delayEach, tap } from 'ix/asynciterable/operators'; | ||
|
|
||
| describe(`AsyncIterable#switchAll`, () => { | ||
| test('switches inner sequences', async () => { | ||
| const innerValues = new Array<string>(); | ||
|
|
||
| const ys = of('0', '1', '2', '3').pipe( | ||
| delayEach(200), | ||
| tap((x) => innerValues.push(x)) | ||
| ); | ||
| const xs = of(ys, ys, ys).pipe(delayEach(500)); | ||
| const source = xs.pipe(switchAll()); | ||
| const expected = [ | ||
| '0', | ||
| '1', // xs=0 | ||
| '0', | ||
| '1', // xs=1 | ||
| '0', | ||
| '1', | ||
| '2', | ||
| '3', // xs=2 | ||
| ]; | ||
|
|
||
| expect(await toArray(source)).toEqual(expected); | ||
|
|
||
| expect(innerValues).toEqual(expected); | ||
| }); | ||
|
|
||
| test(`supports projecting to Arrays`, async () => { | ||
| const xs = of([0, 1, 2], [0, 1, 2], [0, 1, 2]).pipe(delayEach(100)); | ||
| const source = xs.pipe(switchAll()); | ||
| expect(await toArray(source)).toEqual([0, 1, 2, 0, 1, 2, 0, 1, 2]); | ||
| }); | ||
|
|
||
| test(`supports projecting to Promise<Array>`, async () => { | ||
| const xs = of( | ||
| Promise.resolve([0, 1, 2]), | ||
| Promise.resolve([0, 1, 2]), | ||
| Promise.resolve([0, 1, 2]) | ||
| ).pipe(delayEach(100)); | ||
| const source = xs.pipe(switchAll()); | ||
| expect(await toArray(source)).toEqual([0, 1, 2, 0, 1, 2, 0, 1, 2]); | ||
| }); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| import { AsyncIterableX } from '../../asynciterable/asynciterablex'; | ||
| import { switchAll } from '../../asynciterable/operators/switchall'; | ||
|
|
||
| /** | ||
| * @ignore | ||
| */ | ||
| export function switchAllProto<T>(this: AsyncIterableX<AsyncIterable<T>>) { | ||
| return switchAll()(this); | ||
| } | ||
|
|
||
| AsyncIterableX.prototype.switchAll = switchAllProto; | ||
|
|
||
| declare module '../../asynciterable/asynciterablex' { | ||
| interface AsyncIterableX<T> { | ||
| switchAll: typeof switchAllProto; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.