buffer-pipe.ts

import { OperatorAsyncFunction } from '../../interfaces';
import { AsyncIterableX } from '../../asynciterable';
import { BufferAsyncIterable } from '../buffer';

export function buffer<TSource>(
    count: number,
    skip?: number): OperatorAsyncFunction<TSource, TSource[]> {
  if (skip == null) { skip = count; }
  return function bufferCountOperatorFunction(source: AsyncIterable<TSource>): AsyncIterableX<TSource[]> {
    return new BufferAsyncIterable<TSource>(source, count, skip!);
  };
}

interfaces.ts

import { IterableX } from './iterable';
import { AsyncIterableX } from './asynciterable';

export type UnaryFunction<T, R> = (source: T) => R;

export type OperatorFunction<T, R> = UnaryFunction<Iterable<T>, IterableX<R>>;

export type OperatorAsyncFunction<T, R> = UnaryFunction<AsyncIterable<T>, AsyncIterableX<R>>;

export type FactoryOrValue<T> = T | (() => T);

export type MonoTypeOperatorFunction<T> = OperatorFunction<T, T>;

export type MonoTypeOperatorAsyncFunction<T> = OperatorAsyncFunction<T, T>;