ldiamond/strom
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Clean up

Browse Source
This commit is contained in:
Jerry Kurian 2019-09-26 09:23:09 -04:00
parent 70edee51c4
commit a11aa10d16
12 changed files with 36 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/functions/accumulator.ts
View File

@ -1,7 +1,13 @@
import { Transform, TransformOptions } from "stream";
import { AccumulatorByIteratee, FlushStrategy } from "./baseDefinitions";
import { batch } from ".";
export enum FlushStrategy {
rolling = "rolling",
sliding = "sliding",
}
export type AccumulatorByIteratee<T> = (event: T, bufferChunk: T) => boolean;
function _accumulator<T>(
accumulateBy: (data: T, buffer: T[], stream: Transform) => void,
shouldFlush: boolean = true,

15
src/functions/baseDefinitions.ts
View File

@ -1,12 +1,3 @@
export interface ThroughOptions {
objectMode?: boolean;
}
export interface TransformOptions {
readableObjectMode?: boolean;
writableObjectMode?: boolean;
}
export interface WithEncoding {
encoding: string;
}
@ -21,9 +12,3 @@ export type JsonValue = JsonPrimitive | JsonPrimitive[];
export interface JsonParseOptions {
pretty: boolean;
}
export enum FlushStrategy {
rolling = "rolling",
sliding = "sliding",
}
export type AccumulatorByIteratee<T> = (event: T, bufferChunk: T) => boolean;

6
src/functions/batch.ts
View File

@ -1,12 +1,10 @@
import { Transform } from "stream";
import { TransformOptions } from "./baseDefinitions";
import { Transform, TransformOptions } from "stream";
export function batch(
batchSize: number = 1000,
maxBatchAge: number = 500,
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
objectMode: true,
},
): Transform {
let buffer: any[] = [];

7
src/functions/collect.ts
View File

@ -1,9 +1,6 @@
import { Transform } from "stream";
import { ThroughOptions } from "./baseDefinitions";
import { Transform, TransformOptions } from "stream";
export function collect(
options: ThroughOptions = { objectMode: false },
): Transform {
export function collect(options: TransformOptions = {}): Transform {
const collected: any[] = [];
return new Transform({
...options,

6
src/functions/flatMap.ts
View File

@ -1,13 +1,11 @@
import { Transform } from "stream";
import { TransformOptions } from "./baseDefinitions";
import { Transform, TransformOptions } from "stream";
export function flatMap<T, R>(
mapper:
| ((chunk: T, encoding: string) => R[])
| ((chunk: T, encoding: string) => Promise<R[]>),
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
objectMode: true,
},
): Transform {
return new Transform({

8
src/functions/map.ts
View File

@ -1,12 +1,8 @@
import { Transform } from "stream";
import { TransformOptions } from "./baseDefinitions";
import { Transform, TransformOptions } from "stream";
export function map<T, R>(
mapper: (chunk: T, encoding: string) => R,
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
},
options: TransformOptions = { objectMode: true },
): Transform {
return new Transform({
...options,

6
src/functions/parallelMap.ts
View File

@ -1,14 +1,12 @@
import { Transform } from "stream";
import { Transform, TransformOptions } from "stream";
import { sleep } from "../helpers";
import { TransformOptions } from "./baseDefinitions";
export function parallelMap<T, R>(
mapper: (data: T) => R,
parallel: number = 10,
sleepTime: number = 5,
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
objectMode: true,
},
) {
let inflight = 0;

6
src/functions/rate.ts
View File

@ -1,14 +1,12 @@
import { Transform } from "stream";
import { Transform, TransformOptions } from "stream";
import { performance } from "perf_hooks";
import { sleep } from "../helpers";
import { TransformOptions } from "./baseDefinitions";
export function rate(
targetRate: number = 50,
period: number = 1,
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
objectMode: true,
},
): Transform {
const deltaMS = ((1 / targetRate) * 1000) / period; // Skip a full period

9
src/functions/reduce.ts
View File

@ -1,5 +1,4 @@
import { Transform } from "stream";
import { TransformOptions } from "./baseDefinitions";
import { Transform, TransformOptions } from "stream";
export function reduce<T, R>(
iteratee:
@ -7,14 +6,12 @@ export function reduce<T, R>(
| ((previousValue: R, chunk: T, encoding: string) => Promise<R>),
initialValue: R,
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
objectMode: true,
},
) {
let value = initialValue;
return new Transform({
readableObjectMode: options.readableObjectMode,
writableObjectMode: options.writableObjectMode,
...options,
async transform(chunk: T, encoding, callback) {
value = await iteratee(value, chunk, encoding);
callback();

6
src/functions/unbatch.ts
View File

@ -1,10 +1,8 @@
import { Transform } from "stream";
import { TransformOptions } from "./baseDefinitions";
import { Transform, TransformOptions } from "stream";
export function unbatch(
options: TransformOptions = {
readableObjectMode: true,
writableObjectMode: true,
objectMode: true,
},
) {
return new Transform({

2
tests/accumulator.spec.ts
View File

@ -2,7 +2,7 @@ import test from "ava";
import { expect } from "chai";
import { Readable } from "stream";
import { accumulator, accumulatorBy } from "../src";
import { FlushStrategy } from "../src/functions/baseDefinitions";
import { FlushStrategy } from "../src/functions/accumulator";
import { performance } from "perf_hooks";
test.cb("accumulator() rolling", t => {

13
tests/demux.spec.ts
View File

@ -379,6 +379,7 @@ test.cb.only(
const construct = (destKey: string) => {
const first = map(
(chunk: Chunk) => {
console.log("1: ", chunk);
chunk.mapped.push(1);
return chunk;
},
@ -387,7 +388,9 @@ test.cb.only(
const second = map(
async (chunk: Chunk) => {
console.log("2: ", chunk);
await sleep(slowProcessorSpeed);
console.log("2 done ", chunk);
chunk.mapped.push(2);
return chunk;
},
@ -408,6 +411,7 @@ test.cb.only(
// This event should be received after at least 5 * slowProcessorSpeed (two are read immediately by first and second, 5 remaining in demux before drain event)
_demux.on("drain", () => {
expect(_demux._writableState.length).to.be.equal(0);
console.log(performance.now() - start);
expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * (input.length - 2),
);
@ -427,7 +431,7 @@ test.cb.only(
const start = performance.now();
input.forEach(item => {
_demux.write(item);
console.log(_demux.write(item));
});
},
);
@ -457,6 +461,7 @@ test.cb(
const construct = (destKey: string) => {
const first = map(
(chunk: Chunk) => {
console.log("1: ", chunk);
chunk.mapped.push(1);
return chunk;
},
@ -464,6 +469,7 @@ test.cb(
);
const second = map(
(chunk: Chunk) => {
console.log("2: ", chunk);
chunk.mapped.push(2);
return chunk;
},
@ -472,7 +478,9 @@ test.cb(
const third = map(
async (chunk: Chunk) => {
console.log("3: ", chunk);
await sleep(slowProcessorSpeed);
console.log(" 3 done ", chunk);
chunk.mapped.push(3);
return chunk;
},
@ -496,6 +504,7 @@ test.cb(
// This event should be received after at least 3 * slowProcessorSpeed (two are read immediately by first and second, 3 remaining in demux before drain event)
_demux.on("drain", () => {
expect(_demux._writableState.length).to.be.equal(0);
console.log(performance.now() - start);
expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * (input.length - 4),
);
@ -515,7 +524,7 @@ test.cb(
const start = performance.now();
input.forEach(item => {
_demux.write(item);
console.log(_demux.write(item));
});
},
);