diff --git a/package.json b/package.json index 38dc4c5..8009f9f 100644 --- a/package.json +++ b/package.json @@ -31,6 +31,7 @@ "devDependencies": { "@types/chai": "^4.1.7", "@types/node": "^10.12.10", + "@types/typescript": "^2.0.0", "ava": "^1.0.0-rc.2", "chai": "^4.2.0", "mhysa": "./", diff --git a/src/functions/definitions.ts b/src/functions/definitions.ts index f18659a..8abb2ea 100644 --- a/src/functions/definitions.ts +++ b/src/functions/definitions.ts @@ -22,18 +22,3 @@ export interface JsonParseOptions { pretty: boolean; } -export interface IBatchParams { - batchSize?: number; - maxBatchAge?: number; -} - -export interface IRateParams { - targetRate?: number; - period?: number; -} - -export interface IParallelMapParams { - mapper: (data: T) => R; - parallel?: number; - sleepTime?: number; -} diff --git a/src/functions/functions.spec.ts b/src/functions/functions.spec.ts index 50e5916..02e5303 100644 --- a/src/functions/functions.spec.ts +++ b/src/functions/functions.spec.ts @@ -25,7 +25,7 @@ import { rate, parallelMap, } from "."; -import { SerializationFormats } from "./definitions"; +import { sleep } from "../helpers"; test.cb("fromArray() streams array elements in flowing mode", t => { t.plan(3); @@ -683,7 +683,7 @@ test.cb("parse() parses the streamed elements as JSON", t => { const expectedElements = ["abc", {}, []]; let i = 0; source - .pipe(parse(SerializationFormats.utf8)) + .pipe(parse()) .on("data", part => { expect(part).to.deep.equal(expectedElements[i]); t.pass(); @@ -702,7 +702,7 @@ test.cb("parse() emits errors on invalid JSON", t => { t.plan(2); const source = new Readable({ objectMode: true }); source - .pipe(parse(SerializationFormats.utf8)) + .pipe(parse()) .resume() .on("error", () => t.pass()) .on("end", t.end); @@ -1297,7 +1297,7 @@ test.cb("rate() sends data at desired rate", t => { }); test.cb("parallel() parallel mapping", t => { - t.plan(5); + t.plan(6); const source = new Readable({ objectMode: true }); const expectedElements = [ "a_processed", @@ -1305,22 +1305,48 @@ test.cb("parallel() parallel mapping", t => { "c_processed", "d_processed", "e_processed", + "f_processed", ]; - const orderedResults: string[] = []; + const orderedResults: Array<{ output: string; processed: number }> = []; // Record start / end times of each process and then compare to figure out # of processes ocurring and order source - .pipe(parallelMap(data => data + "_processed")) + .pipe( + parallelMap(async (data: any) => { + const c = data + "_processed"; + await sleep(500); + orderedResults.push({ + output: c, + processed: performance.now(), + }); + return c; + }, 2), + ) .on("data", (element: string) => { t.true(expectedElements.includes(element)); - orderedResults.push(element); }) .on("error", t.end) - .on("end", () => { - expect(orderedResults[0]).to.equal("a_processed"); - expect(orderedResults[1]).to.equal("b_processed"); - expect(orderedResults[2]).to.equal("d_processed"); - expect(orderedResults[3]).to.equal("c_processed"); - expect(orderedResults[4]).to.equal("e_processed"); + .on("end", async () => { + expect(orderedResults[0].processed).to.be.lessThan( + orderedResults[1].processed + 500, + ); + expect(orderedResults[2].processed).to.be.lessThan( + orderedResults[3].processed + 500, + ); + expect(orderedResults[4].processed).to.be.lessThan( + orderedResults[5].processed + 500, + ); + expect(orderedResults[2].processed).to.be.greaterThan( + orderedResults[0].processed + 500, + ); + expect(orderedResults[3].processed).to.be.greaterThan( + orderedResults[1].processed + 500, + ); + expect(orderedResults[4].processed).to.be.greaterThan( + orderedResults[2].processed + 500, + ); + expect(orderedResults[5].processed).to.be.greaterThan( + orderedResults[3].processed + 500, + ); t.end(); }); @@ -1329,5 +1355,6 @@ test.cb("parallel() parallel mapping", t => { source.push("c"); source.push("d"); source.push("e"); + source.push("f"); source.push(null); }); diff --git a/src/functions/index.ts b/src/functions/index.ts index 1ad0f10..9085930 100644 --- a/src/functions/index.ts +++ b/src/functions/index.ts @@ -7,9 +7,6 @@ import { TransformOptions, WithEncoding, JsonParseOptions, - IBatchParams, - IRateParams, - IParallelMapParams, } from "./definitions"; /** @@ -205,14 +202,10 @@ export function last(readable: Readable): Promise { /** * Stores chunks of data internally in array and batches when batchSize is reached. - * - * @param batchSize? Size of the batches, defaults to 1000. + * @param batchSize Size of the batches, defaults to 1000. * @param maxBatchAge? Max lifetime of a batch, defaults to 500 */ -export function batch({ - batchSize, - maxBatchAge, -}: IBatchParams): NodeJS.ReadWriteStream { +export function batch(batchSize: number, maxBatchAge?: number): NodeJS.ReadWriteStream { return baseFunctions.batch(batchSize, maxBatchAge); } @@ -225,13 +218,11 @@ export function unbatch(): NodeJS.ReadWriteStream { /** * Limits date of data transferred into stream. + * @param options? * @param targetRate? Desired rate in ms * @param period? Period to sleep for when rate is above or equal to targetRate */ -export function rate({ - targetRate, - period, -}: IRateParams): NodeJS.ReadWriteStream { +export function rate(targetRate?: number, period?: number): NodeJS.ReadWriteStream { return baseFunctions.rate(targetRate, period); } @@ -241,10 +232,10 @@ export function rate({ * @param func Function to execute on each data chunk * @param pause Amount of time to pause processing when max number of parallel processes are executing. */ -export function parallelMap({ - mapper, - parallel, - sleepTime, -}: IParallelMapParams) { +export function parallelMap( + mapper: (chunk: T) => R, + parallel?: number, + sleepTime?: number, +) { return baseFunctions.parallelMap(mapper, parallel, sleepTime); } diff --git a/yarn.lock b/yarn.lock index 082e22d..dfb9135 100644 --- a/yarn.lock +++ b/yarn.lock @@ -335,6 +335,13 @@ resolved "https://registry.yarnpkg.com/@types/node/-/node-10.12.10.tgz#4fa76e6598b7de3f0cb6ec3abacc4f59e5b3a2ce" integrity sha512-8xZEYckCbUVgK8Eg7lf5Iy4COKJ5uXlnIOnePN0WUwSQggy9tolM+tDJf7wMOnT/JT/W9xDYIaYggt3mRV2O5w== +"@types/typescript@^2.0.0": + version "2.0.0" + resolved "https://registry.yarnpkg.com/@types/typescript/-/typescript-2.0.0.tgz#c433539c98bae28682b307eaa7a0fd2115b83c28" + integrity sha1-xDNTnJi64oaCswfqp6D9IRW4PCg= + dependencies: + typescript "*" + abbrev@1: version "1.1.1" resolved "https://registry.yarnpkg.com/abbrev/-/abbrev-1.1.1.tgz#f8f2c887ad10bf67f634f005b6987fed3179aac8" @@ -3205,6 +3212,11 @@ type-detect@^4.0.0, type-detect@^4.0.5: resolved "https://registry.yarnpkg.com/type-detect/-/type-detect-4.0.8.tgz#7646fb5f18871cfbb7749e69bd39a6388eb7450c" integrity sha512-0fr/mIH1dlO+x7TlcMy+bIDqKPsw/70tVyeHW787goQjhmqaZe10uwLujubK9q9Lg6Fiho1KUKDYz0Z7k7g5/g== +typescript@*: + version "3.5.2" + resolved "https://registry.yarnpkg.com/typescript/-/typescript-3.5.2.tgz#a09e1dc69bc9551cadf17dba10ee42cf55e5d56c" + integrity sha512-7KxJovlYhTX5RaRbUdkAXN1KUZ8PwWlTzQdHV6xNqvuFOs7+WBo10TQUqT19Q/Jz2hk5v9TQDIhyLhhJY4p5AA== + typescript@^3.1.6: version "3.1.6" resolved "https://registry.yarnpkg.com/typescript/-/typescript-3.1.6.tgz#b6543a83cfc8c2befb3f4c8fba6896f5b0c9be68"