Default to flatMap concurrent (#347)

* feat(flat): default AsyncIterable#flat's concurrent parameter to Infinity

* feat(mergeall): default AsyncIterable#mergeAll's concurrent parameter to Infinity

* feat(flatmap): default AsyncIterable#flatMap's concurrent parameter to Infinity

spec/asynciterable-operators/flat-spec.ts

@@ -6,16 +6,16 @@ function compareArrays<T>(fst: Iterable<T>, snd: Iterable<T>) {
   expect(fst.toString()).toBe(snd.toString());
 }
 
-test('AsyncIterable#flat flattens all', async () => {
+test('AsyncIterable#flat flattens all layers', async () => {
   const xs = of(1, of(2, of(3)), 4);
   const ys = await toArray(xs.pipe(flat()));
 
-  compareArrays(ys, [1, 2, 3, 4]);
+  compareArrays(ys, [1, 4, 2, 3]);
 });
 
-test('AsyncIterable#flat flattens all layers', async () => {
+test('AsyncIterable#flat flattens all layers with concurrent = 1', async () => {
   const xs = of(1, of(2, of(3)), 4);
-  const ys = await toArray(xs.pipe(flat(-1)));
+  const ys = await toArray(xs.pipe(flat(-1, 1)));
 
   compareArrays(ys, [1, 2, 3, 4]);
 });
@@ -24,5 +24,5 @@ test('AsyncIterable#flat flattens two layers', async () => {
   const xs = of(1, of(2, of(3)), 4);
   const ys = await toArray(xs.pipe(flat(2)));
 
-  compareArrays(ys, [1, 2, 3, 4]);
+  compareArrays(ys, [1, 4, 2, 3]);
 });

spec/asynciterable-operators/flatmap-spec.ts

@@ -1,19 +1,12 @@
-import { hasNext, noNext } from '../asynciterablehelpers';
-import { of, range, throwError } from 'ix/asynciterable';
+import { hasNext } from '../asynciterablehelpers';
+import { of, range, throwError, toArray } from 'ix/asynciterable';
 import { flatMap } from 'ix/asynciterable/operators';
 
 test('AsyncIterable#flatMap with range', async () => {
   const xs = of(1, 2, 3);
   const ys = xs.pipe(flatMap(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);
+  expect(await toArray(ys)).toEqual([0, 0, 0, 1, 1, 2]);
 });
 
 test('AsyncIterable#flatMap selector returns throw', async () => {

spec/asynciterable-operators/mergeall-spec.ts

@@ -4,5 +4,5 @@ import { mergeAll } from 'ix/asynciterable/operators';
 
 test('AsyncIterable#merge mergeAll behavior', async () => {
   const res = of(of(1, 2, 3), of(4, 5)).pipe(mergeAll());
-  expect(await toArray(res)).toEqual([1, 2, 3, 4, 5]);
+  expect(await toArray(res)).toEqual([1, 2, 4, 3, 5]);
 });

src/add/asynciterable-operators/flat.ts

@@ -4,8 +4,12 @@ import { flat } from '../../asynciterable/operators/flat';
 /**
  * @ignore
  */
-export function flatProto<T, D extends number = -1>(this: AsyncIterableX<T>, depth: D = -1 as any) {
-  return flat(depth)(this);
+export function flatProto<T, D extends number = -1>(
+  this: AsyncIterableX<T>,
+  depth: D = -1 as any,
+  concurrent = Infinity
+) {
+  return flat(depth, concurrent)(this);
 }
 
 AsyncIterableX.prototype.flat = flatProto;

src/add/asynciterable-operators/flatmap.ts

@@ -8,9 +8,10 @@ import { FlattenConcurrentSelector } from '../../asynciterable/operators/_flatte
 export function flatMapProto<T, R>(
   this: AsyncIterableX<T>,
   selector: FlattenConcurrentSelector<T, R>,
+  concurrent = Infinity,
   thisArg?: any
 ) {
-  return flatMap(selector, thisArg)(this);
+  return flatMap(selector, concurrent, thisArg)(this);
 }
 
 AsyncIterableX.prototype.flatMap = flatMapProto;

src/add/asynciterable-operators/mergeall.ts

@@ -4,8 +4,11 @@ import { mergeAll } from '../../asynciterable/operators/mergeall';
 /**
  * @ignore
  */
-export function mergeAllProto<T>(this: AsyncIterableX<AsyncIterable<T>>): AsyncIterableX<T> {
-  return mergeAll()(this);
+export function mergeAllProto<T>(
+  this: AsyncIterableX<AsyncIterable<T>>,
+  concurrent = Infinity
+): AsyncIterableX<T> {
+  return mergeAll(concurrent)(this);
 }
 
 AsyncIterableX.prototype.mergeAll = mergeAllProto;

src/asynciterable/operators/flat.ts

@@ -25,7 +25,7 @@ type FlattenWithDepth<Arr, Depth extends number> = {
  * @param {number} [depth=Infinity] The depth to flatten the async-iterable sequence if specified, otherwise infinite.
  * @returns {MonoTypeOperatorAsyncFunction<T>} An operator that flattens the async-iterable sequence.
  */
-export function flat<D extends number = -1>(depth: D = -1 as any) {
+export function flat<D extends number = -1>(depth: D = -1 as any, concurrent = Infinity) {
   depth = (depth < 0 ? Infinity : depth) as any;
   return function flattenOperatorFunction<T>(
     source: AsyncIterable<T>
@@ -35,6 +35,6 @@ export function flat<D extends number = -1>(depth: D = -1 as any) {
         return depth > 0 ? flat(depth - 1)(item) : item;
       }
       return [item];
-    })(source) as AsyncIterableX<Flattened<T, D>>;
+    }, concurrent)(source) as AsyncIterableX<Flattened<T, D>>;
   };
 }

src/asynciterable/operators/flatmap.ts

@@ -18,9 +18,10 @@ import { OperatorAsyncFunction } from '../../interfaces';
  */
 export function flatMap<TSource, TResult>(
   selector: FlattenConcurrentSelector<TSource, TResult>,
+  concurrent = Infinity,
   thisArg?: any
 ): OperatorAsyncFunction<TSource, TResult> {
   return function flatMapOperatorFunction(source) {
-    return new FlattenConcurrentAsyncIterable(source, selector, 1, false, thisArg);
+    return new FlattenConcurrentAsyncIterable(source, selector, concurrent, false, thisArg);
   };
 }

src/asynciterable/operators/mergeall.ts

@@ -6,8 +6,8 @@ import { FlattenConcurrentAsyncIterable } from './_flatten';
  * @template TSource The type of the elements in the source sequences.
  * @returns {OperatorAsyncFunction<AsyncIterable<TSource>, TSource>} The async-iterable sequence that merges the elements of the inner sequences.
  */
-export function mergeAll() {
+export function mergeAll(concurrent = Infinity) {
   return function mergeAllOperatorFunction<TSource>(source: AsyncIterable<AsyncIterable<TSource>>) {
-    return new FlattenConcurrentAsyncIterable(source, (s) => s, 1, false);
+    return new FlattenConcurrentAsyncIterable(source, (s) => s, concurrent, false);
   };
 }