Merge pull request #24 from scramjetorg/feature/reduce

Introduce "DataStream.reduce()"

Author: f1ames

src/streams/base-stream.ts

@@ -1,6 +1,7 @@
-import { TransformFunction, AnyIterable } from "../types";
+import { TransformFunction, AnyIterable, MaybePromise } from "../types";
 export interface BaseStream<T extends any> {
     map<U, W extends any[]>(callback: TransformFunction<T, U, W>, ...args: W): BaseStream<U>;
     flatMap<U, W extends any[]>(callback: TransformFunction<T, AnyIterable<U>, W>, ...args: W): BaseStream<U>;
     filter<W extends any[]>(callback: TransformFunction<T, Boolean, W>, ...args: W): BaseStream<T>;
+    reduce<U = T>(callback: (previousValue: U, currentChunk: T) => MaybePromise<U>, initial?: U): Promise<U>
 }

src/streams/data-stream.ts

@@ -2,9 +2,16 @@ import { Readable } from "stream";
 import { createReadStream, promises as fs } from "fs";
 import { BaseStream } from "./base-stream";
 import { IFCA } from "../ifca";
-import { AnyIterable, Constructor, DroppedChunk, ResolvablePromiseObject, TransformFunction } from "../types";
+import { AnyIterable, Constructor, DroppedChunk, ResolvablePromiseObject, TransformFunction, MaybePromise } from "../types";
 import { createResolvablePromiseObject, isAsyncFunction } from "../utils";
 
+type Reducer<T, U> = {
+    isAsync: boolean,
+    value?: U,
+    onFirstChunkCallback: Function,
+    onChunkCallback: (chunk: T) => MaybePromise<void>
+};
+
 export class DataStream<T> implements BaseStream<T>, AsyncIterable<T> {
     constructor() {
         this.ifca = new IFCA<T, T, any>(2, (chunk: T) => chunk);
@@ -67,10 +74,27 @@ export class DataStream<T> implements BaseStream<T>, AsyncIterable<T> {
         return this.asNewFlattenedStream(this.map<AnyIterable<U>, W>(callback, ...args));
     }
 
+    async reduce<U = T>(callback: (previousValue: U, currentChunk: T) => MaybePromise<U>, initial?: U): Promise<U> {
+        // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/Reduce#parameters
+        //
+        // initialValue (optional):
+        // A value to which previousValue is initialized the first time the callback is called.
+        // If initialValue is specified, that also causes currentValue to be initialized to the first
+        // value in the array. If initialValue is not specified, previousValue is initialized to the first
+        // value in the array, and currentValue is initialized to the second value in the array.
+
+        const reducer = this.getReducer<U>(callback, initial);
+        const reader = reducer.isAsync
+            ? this.getReaderAsyncCallback(true, reducer)
+            : this.getReader(true, reducer);
+
+        return reader().then(() => reducer.value as U);
+    }
+
     async toArray(): Promise<T[]> {
         const chunks: Array<T> = [];
 
-        await (this.getReader(true, chunk => { chunks.push(chunk); }))();
+        await (this.getReader(true, { onChunkCallback: chunk => { chunks.push(chunk); } }))();
 
         return chunks;
     }
@@ -97,6 +121,37 @@ export class DataStream<T> implements BaseStream<T>, AsyncIterable<T> {
         }
     }
 
+    protected getReducer<U>(
+        callback: (previousValue: U, currentChunk: T) => MaybePromise<U>,
+        initial?: U
+    ): Reducer<T, U> {
+        const reducer: any = {
+            isAsync: isAsyncFunction(callback),
+            value: initial
+        };
+
+        reducer.onFirstChunkCallback = async (chunk: T): Promise<void> => {
+            if (initial === undefined) {
+                // Here we should probably check if typeof chunk is U.
+                reducer.value = chunk as unknown as U;
+            } else {
+                reducer.value = await callback(reducer.value as U, chunk);
+            }
+        };
+
+        if (reducer.isAsync) {
+            reducer.onChunkCallback = async (chunk: T): Promise<void> => {
+                reducer.value = await callback(reducer.value as U, chunk) as U;
+            };
+        } else {
+            reducer.onChunkCallback = (chunk: T): void => {
+                reducer.value = callback(reducer.value as U, chunk) as U;
+            };
+        }
+
+        return reducer as Reducer<T, U>;
+    }
+
     protected asNewFlattenedStream<U, W extends DataStream<AnyIterable<U>>>(
         fromStream: W,
         onEndYield?: () => { yield: boolean, value?: U }
@@ -116,34 +171,109 @@ export class DataStream<T> implements BaseStream<T>, AsyncIterable<T> {
         })(fromStream));
     }
 
-    // For now this method assumes both callbacks are sync ones.
     protected getReader(
         uncork: boolean,
-        onChunkCallback: (chunk: T) => void,
-        onEndCallback?: Function
+        callbacks: {
+            onChunkCallback: (chunk: T) => void,
+            onFirstChunkCallback?: Function,
+            onEndCallback?: Function
+        }
     ): () => Promise<void> {
+        /* eslint-disable complexity */
         return async () => {
             if (uncork && this.corked) {
                 this._uncork();
             }
 
+            let chunk = this.ifca.read();
+
+            // A bit of code duplication but we don't want to have unnecessary if inside a while loop
+            // which is called for every chunk or wrap the common code inside another function due to performance.
+            if (callbacks.onFirstChunkCallback) {
+                if (chunk instanceof Promise) {
+                    chunk = await chunk;
+                }
+
+                if (chunk !== null) {
+                    await callbacks.onFirstChunkCallback(chunk);
+                    chunk = this.ifca.read();
+                }
+            }
+
             // eslint-disable-next-line no-constant-condition
             while (true) {
-                let chunk = this.ifca.read();
+                if (chunk instanceof Promise) {
+                    chunk = await chunk;
+                }
+
+                if (chunk === null) {
+                    break;
+                }
+
+                callbacks.onChunkCallback(chunk);
+
+                chunk = this.ifca.read();
+            }
+
+            if (callbacks.onEndCallback) {
+                await callbacks.onEndCallback.call(this);
+            }
+        };
+        /* eslint-enable complexity */
+    }
+
+    // This is duplicated '.getReader()' method with the only difference that 'onChunkCallback'
+    // is an async function so we have to 'await' on it for each chunk. Since it has significant effect
+    // on processing time (and makes it asynchronous) I have extracted it as a separate method.
+    protected getReaderAsyncCallback(
+        uncork: boolean,
+        callbacks: {
+            onChunkCallback: (chunk: T) => MaybePromise<void>,
+            onFirstChunkCallback?: Function,
+            onEndCallback?: Function
+        }
+    ): () => Promise<void> {
+        /* eslint-disable complexity */
+        return async () => {
+            if (uncork && this.corked) {
+                this._uncork();
+            }
+
+            let chunk = this.ifca.read();
 
+            // A bit of code duplication but we don't want to have unnecessary if inside a while loop
+            // which is called for every chunk or wrap the common code inside another function due to performance.
+            if (callbacks.onFirstChunkCallback) {
                 if (chunk instanceof Promise) {
                     chunk = await chunk;
                 }
+
+                if (chunk !== null) {
+                    await callbacks.onFirstChunkCallback(chunk);
+                    chunk = this.ifca.read();
+                }
+            }
+
+            // eslint-disable-next-line no-constant-condition
+            while (true) {
+                if (chunk instanceof Promise) {
+                    chunk = await chunk;
+                }
+
                 if (chunk === null) {
-                    if (onEndCallback) {
-                        onEndCallback.call(this);
-                    }
                     break;
                 }
 
-                onChunkCallback(chunk);
+                await callbacks.onChunkCallback(chunk);
+
+                chunk = this.ifca.read();
+            }
+
+            if (callbacks.onEndCallback) {
+                await callbacks.onEndCallback.call(this);
             }
         };
+        /* eslint-enable complexity */
     }
 
     // Native node readables also implement AsyncIterable interface.

src/utils.ts

@@ -27,19 +27,19 @@ function createResolvablePromiseObject<T>(): ResolvablePromiseObject<T> {
     return { promise, resolver: resolver as () => (T) };
 }
 
-function isIterable(iterable: any): Boolean {
+function isIterable(iterable: any): boolean {
     return iterable && iterable[Symbol.iterator] && typeof iterable[Symbol.iterator] === "function";
 }
 
-function isAsyncIterable(iterable: any): Boolean {
+function isAsyncIterable(iterable: any): boolean {
     return iterable && iterable[Symbol.asyncIterator] && typeof iterable[Symbol.asyncIterator] === "function";
 }
 
-function isAsyncFunction(func: any): Boolean {
+function isAsyncFunction(func: any): boolean {
     return func && func[Symbol.toStringTag] === "AsyncFunction";
 }
 
-function isAsyncTransformHandler(func: TransformHandler<any, any>): Boolean {
+function isAsyncTransformHandler(func: TransformHandler<any, any>): boolean {
     return isAsyncFunction(func[0]) || isAsyncFunction(func[1]);
 }
 

test/helpers/utils.ts

@@ -12,6 +12,10 @@ async function defer<X extends any | undefined>(ts: number, out?: X): Promise<X
     return new Promise((res) => setTimeout(() => res(out), ts));
 }
 
+async function deferReturn<X extends any>(ts: number, out: X): Promise<X> {
+    return new Promise((res) => setTimeout(() => res(out), ts));
+}
+
 function writeInput(ifca: IFCA<any, any, any>, input: any[]): void {
     for (const i of input) {
         ifca.write(i);
@@ -58,6 +62,7 @@ const transforms = {
 
 export {
     defer,
+    deferReturn,
     writeInput,
     readNTimes,
     readNTimesConcurrently,

test/streams/data/reduce.spec.ts

@@ -0,0 +1,96 @@
+import test from "ava";
+import { DataStream } from "../../../src/streams/data-stream";
+import { deferReturn } from "../../helpers/utils";
+
+test("DataStream reduce can be use to calculate sum", async (t) => {
+    const result = await DataStream
+        .from([1, 2, 3, 4, 0, 0, 20, 10, 2, 2])
+        .reduce((a, b) => a + b);
+
+    t.deepEqual(result, 44);
+});
+
+test("DataStream reduce can be use to calculate sum (initial provided)", async (t) => {
+    const result = await DataStream
+        .from([1, 2, 3, 4, 0, 0, 20, 10, 2, 2])
+        .reduce((a, b) => a + b, 0);
+
+    t.deepEqual(result, 44);
+});
+
+test("DataStream reduce can be use to concate numbers to string", async (t) => {
+    const result = await DataStream
+        .from([1, 2, 3, 4, 5, 6, 7, 8, 9, 0])
+        .reduce((a, b) => `${a}${b}`, "");
+
+    t.deepEqual(result, "1234567890");
+});
+
+test("DataStream reduce can be use to calculate sum (async)", async (t) => {
+    const result = await DataStream
+        .from([1, 2, 3, 4, 0, 0, 20, 10, 2, 2])
+        .reduce(async (a, b) => deferReturn(5, a + b));
+
+    t.deepEqual(result, 44);
+});
+
+test("DataStream reduce can be use to calculate sum (initial provided, async)", async (t) => {
+    const result = await DataStream
+        .from([1, 2, 3, 4, 0, 0, 20, 10, 2, 2])
+        .reduce(async (a, b) => deferReturn(5, a + b), 0);
+
+    t.deepEqual(result, 44);
+});
+
+test("DataStream reduce can be use to concate numbers to string (async)", async (t) => {
+    const result = await DataStream
+        .from([1, 2, 3, 4, 5, 6, 7, 8, 9, 0])
+        .reduce(async (a, b) => deferReturn(5, `${a}${b}`), "");
+
+    t.deepEqual(result, "1234567890");
+});
+
+// Tests below inspired by:
+// https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/Reduce#examples
+
+test("DataStream can sum values in an object stream", async (t) => {
+    const result = await DataStream
+        .from([{ x: 1 }, { x: 2 }, { x: 3 }])
+        .reduce((a, b) => a + b.x, 0);
+
+    t.deepEqual(result, 6);
+});
+
+test("DataStream can flatten a stream of arrays (explicit inital)", async (t) => {
+    const initial: number[] = [];
+    const result = await DataStream
+        .from([[0, 1], [2, 3], [4, 5]])
+        .reduce((a, b) => a.concat(b), initial);
+
+    t.deepEqual(result, [0, 1, 2, 3, 4, 5]);
+});
+
+test("DataStream can flatten a stream of arrays (explicit type)", async (t) => {
+    const result = await DataStream
+        .from([[0, 1], [2, 3], [4, 5]])
+        .reduce<number[]>((a, b) => a.concat(b), []);
+
+    t.deepEqual(result, [0, 1, 2, 3, 4, 5]);
+});
+
+test("DataStream can count instances of values in an object stream", async (t) => {
+    const initial: any = {};
+    const result = await DataStream
+        .from(["Alice", "Bob", "Tiff", "Bruce", "Alice"])
+        .reduce((allNames, name) => {
+            if (name in allNames) {
+                allNames[name]++;
+            } else {
+                allNames[name] = 1;
+            }
+
+            return allNames;
+        }, initial);
+
+    t.deepEqual(result, { Alice: 2, Bob: 1, Tiff: 1, Bruce: 1 });
+});