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

Fix tests for Node 15

Browse Source
This commit is contained in:
Lewis Diamond 2021-02-05 22:31:16 -05:00
parent 0e703d0778
commit 4b6c31b9c1
11 changed files with 861 additions and 1666 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
package.json
View File

@ -13,9 +13,9 @@
"url": "https://github.com/Wenzil"
},
{
"name": "Jerry Kurian",
"email": "jerrykurian@protonmail.com",
"url": "https://github.com/jkurian"
"name": "Jerry Kurian",
"email": "jerrykurian@protonmail.com",
"url": "https://github.com/jkurian"
},
{
"name": "Lewis Diamond",
@ -42,29 +42,26 @@
},
"dependencies": {},
"devDependencies": {
"@types/chai": "^4.1.7",
"@types/node": "^12.12.15",
"@types/sinon": "^7.0.13",
"ava": "^2.4.0",
"chai": "^4.2.0",
"@ava/typescript": "^1.1.1",
"@types/chai": "^4.2.14",
"@types/node": "^14.14.25",
"@types/sinon": "^9.0.10",
"ava": "^3.15.0",
"chai": "^4.3.0",
"prettier": "^2.2.1",
"sinon": "^9.2.4",
"stromjs": "./",
"prettier": "^1.14.3",
"sinon": "^7.4.2",
"ts-node": "^8.3.0",
"tslint": "^5.11.0",
"ts-node": "^9.1.1",
"tslint": "^6.1.3",
"tslint-config-prettier": "^1.16.0",
"tslint-plugin-prettier": "^2.0.1",
"typescript": "^3.5.3"
"tslint-plugin-prettier": "^2.3.0",
"typescript": "^4.1.3"
},
"ava": {
"files": [
"tests/*.spec.ts",
"tests/utils/*.spec.ts"
],
"sources": [
"src/**/*.ts"
],
"compileEnhancements": false,
"failWithoutAssertions": false,
"extensions": [
"ts"

2
samples/parallelMap.js
View File

@ -8,7 +8,7 @@ strom
.fromArray([1, 2, 3, 4, 6, 8])
.pipe(
strom.parallelMap(async d => {
await sleep(10000 - d * 1000);
await sleep(1000 - d * 100);
return `${d}`;
}, 3),
)

2
src/functions/baseDefinitions.ts
View File

@ -1,5 +1,5 @@
export interface WithEncoding {
encoding: string;
encoding: BufferEncoding;
}
export enum SerializationFormats {

46
src/functions/compose.ts
View File

@ -1,10 +1,8 @@
import { AllStreams, isReadable } from "../helpers";
import { PassThrough, pipeline, TransformOptions, Transform } from "stream";
import { isReadable, isWritable } from "../helpers";
import { PassThrough, pipeline, TransformOptions, Transform, Stream, Readable, Writable } from "stream";
export function compose(
streams: Array<
NodeJS.ReadableStream | NodeJS.ReadWriteStream | NodeJS.WritableStream
>,
streams: (Readable | Writable)[],
errorCallback?: (err: any) => void,
options?: TransformOptions,
): Compose {
@ -15,30 +13,30 @@ export function compose(
return new Compose(streams, errorCallback, options);
}
enum EventSubscription {
Last = 0,
First,
All,
Self,
}
export class Compose extends Transform {
private first: AllStreams;
private last: AllStreams;
private streams: AllStreams[];
private inputStream: ReadableStream;
private first: Writable & Readable;
private last: Readable;
private streams: Stream[];
constructor(
streams: AllStreams[],
streams: (Readable | Writable)[],
errorCallback?: (err: any) => void,
options?: TransformOptions,
) {
super(options);
this.first = new PassThrough(options);
this.last = streams[streams.length - 1];
const last = streams.pop();
if (last && isReadable(last)) {
this.last = last;
} else {
throw new TypeError("Invalid last stream provided, it must be readable");
}
this.streams = streams;
pipeline(
[this.first, ...streams],
[this.first,
...streams,
this.last],
errorCallback ||
((error: any) => {
if (error) {
@ -48,10 +46,10 @@ export class Compose extends Transform {
);
if (isReadable(this.last)) {
(this.last as NodeJS.ReadWriteStream).pipe(
this.last.pipe(
new Transform({
...options,
transform: (d: any, encoding, cb) => {
transform: (d: any, _encoding, cb) => {
this.push(d);
cb();
},
@ -60,13 +58,13 @@ export class Compose extends Transform {
}
}
public _transform(chunk: any, encoding: string, cb: any) {
(this.first as NodeJS.WritableStream).write(chunk, encoding, cb);
public _transform(chunk: any, encoding: BufferEncoding, cb: any) {
this.first.write(chunk, encoding, cb);
}
public _flush(cb: any) {
if (isReadable(this.first)) {
(this.first as any).push(null);
if (isWritable(this.first)) {
this.first.push(null);
}
this.last.once("end", () => {
cb();

22
src/functions/demux.ts
View File

@ -1,9 +1,7 @@
import { DuplexOptions, Duplex, Transform } from "stream";
import { DuplexOptions, Duplex, Transform, Writable } from "stream";
import { isReadable } from "../helpers";
type DemuxStreams = NodeJS.WritableStream | NodeJS.ReadWriteStream;
export interface DemuxOptions extends DuplexOptions {
remultiplex?: boolean;
}
@ -12,7 +10,7 @@ export function demux(
pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams | DemuxStreams[],
) => Writable | Writable[],
demuxBy: string | ((chunk: any) => string),
options?: DemuxOptions,
): Duplex {
@ -21,20 +19,20 @@ export function demux(
class Demux extends Duplex {
private streamsByKey: {
[key: string]: DemuxStreams[];
[key: string]: Writable[];
};
private demuxer: (chunk: any) => string;
private pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams[];
) => Writable[];
private remultiplex: boolean;
private transform: Transform;
constructor(
pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams | DemuxStreams[],
) => Writable | Writable[],
demuxBy: string | ((chunk: any) => string),
options: DemuxOptions = {},
) {
@ -60,7 +58,7 @@ class Demux extends Duplex {
}
// tslint:disable-next-line
public _read(size: number) {}
public _read(_size: number) {}
public async _write(chunk: any, encoding: any, cb: any) {
const destKey = this.demuxer(chunk);
@ -70,7 +68,7 @@ class Demux extends Duplex {
newPipelines.forEach(newPipeline => {
if (this.remultiplex && isReadable(newPipeline)) {
(newPipeline as NodeJS.ReadWriteStream).pipe(
newPipeline.pipe(
this.transform,
);
} else if (this.remultiplex) {
@ -88,7 +86,7 @@ class Demux extends Duplex {
pendingDrains.push(
new Promise(resolve => {
pipeline.once("drain", () => {
resolve();
resolve(null);
});
}),
);
@ -99,7 +97,7 @@ class Demux extends Duplex {
}
public _flush() {
const pipelines: DemuxStreams[] = Array.prototype.concat.apply(
const pipelines: Writable[] = Array.prototype.concat.apply(
[],
Object.values(this.streamsByKey),
);
@ -121,7 +119,7 @@ class Demux extends Duplex {
}
public _destroy(error: any, cb: (error?: any) => void) {
const pipelines: DemuxStreams[] = [].concat.apply(
const pipelines: Writable[] = [].concat.apply(
[],
Object.values(this.streamsByKey),
);

7
src/functions/index.ts
View File

@ -1,9 +1,4 @@
import {
Transform,
TransformOptions,
WritableOptions,
ReadableOptions,
} from "stream";
import { TransformOptions } from "stream";
import { accumulator, accumulatorBy } from "./accumulator";
import { batch } from "./batch";
import { child } from "./child";

2
src/functions/replace.ts
View File

@ -9,7 +9,7 @@ export function replace(
const decoder = new StringDecoder(options.encoding);
return new Transform({
readableObjectMode: true,
transform(chunk: Buffer, encoding, callback) {
transform(chunk: Buffer, _encoding, callback) {
const asString = decoder.write(chunk);
// Take care not to break up multi-byte characters spanning multiple chunks
if (asString !== "" || chunk.length === 0) {

24
src/helpers.ts
View File

@ -1,17 +1,23 @@
import { Readable, Stream, Writable } from "stream";
export async function sleep(time: number): Promise<{} | null> {
return time > 0 ? new Promise(resolve => setTimeout(resolve, time)) : null;
}
export type AllStreams =
| NodeJS.ReadableStream
| NodeJS.ReadWriteStream
| NodeJS.WritableStream;
export function isReadable(
stream: AllStreams,
): stream is NodeJS.WritableStream {
stream: Stream,
): stream is Readable {
return (
(stream as NodeJS.ReadableStream).pipe !== undefined &&
(stream as any).readable === true
(stream as Readable).pipe !== undefined &&
(stream as Readable).readable === true
);
}
export function isWritable(
stream: Stream,
): stream is Writable {
return (
(stream as Writable).write !== undefined &&
(stream as Writable).writable === true
);
}

4
tests/compose.spec.ts
View File

@ -144,7 +144,7 @@ test("compose() writable length should be less than highWaterMark when handing w
});
test("compose() should emit drain event ~rate * highWaterMark ms for every write that causes backpressure", async t => {
t.plan(2);
t.plan(7);
const _rate = 100;
const highWaterMark = 2;
return new Promise(async (resolve, reject) => {
@ -529,10 +529,10 @@ test.cb("compose() `finish` and `end` propagates", t => {
});
composed.on("end", () => {
t.pass();
t.end();
});
sink.on("finish", () => {
t.pass();
t.end();
});
const input = [

42
tests/demux.spec.ts
View File

@ -21,13 +21,11 @@ test.cb("demux() constructor should be called once per key", t => {
{ key: "a", visited: [] },
{ key: "b", visited: [] },
];
const construct = sinon.spy((destKey: string) => {
const dest = map((chunk: Test) => {
const construct = sinon.spy((_destKey: string) => {
return map((chunk: Test) => {
chunk.visited.push(1);
return chunk;
});
return dest;
});
const demuxed = demux(construct, "key", {});
@ -124,13 +122,11 @@ test.cb("demux() constructor should be called once per key using keyBy", t => {
{ key: "b", visited: [] },
];
const construct = sinon.spy((destKey: string) => {
const dest = map((chunk: Test) => {
const construct = sinon.spy((_destKey: string) => {
return map((chunk: Test) => {
chunk.visited.push(1);
return chunk;
});
return dest;
});
const demuxed = demux(construct, item => item.key, {});
@ -206,7 +202,7 @@ test("demux() write should return false and emit drain if more than highWaterMar
let pendingReads = input.length;
const highWaterMark = 5;
const slowProcessorSpeed = 25;
const construct = (destKey: string) => {
const construct = (_destKey: string) => {
const first = map(
async (chunk: Chunk) => {
await sleep(slowProcessorSpeed);
@ -231,7 +227,7 @@ test("demux() write should return false and emit drain if more than highWaterMar
highWaterMark,
});
_demux.on("error", err => {
_demux.on("error", _err => {
reject();
});
@ -239,7 +235,7 @@ test("demux() write should return false and emit drain if more than highWaterMar
const res = _demux.write(item);
expect(_demux._writableState.length).to.be.at.most(highWaterMark);
if (!res) {
await new Promise((resolv, rej) => {
await new Promise((resolv, _rej) => {
_demux.once("drain", () => {
expect(_demux._writableState.length).to.be.equal(0);
t.pass();
@ -270,7 +266,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
const highWaterMark = 5;
const slowProcessorSpeed = 25;
const construct = (destKey: string) => {
const construct = (_destKey: string) => {
const first = map(
async (chunk: Chunk) => {
await sleep(slowProcessorSpeed);
@ -292,7 +288,7 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
const _demux = demux(construct, "key", {
highWaterMark,
});
_demux.on("error", err => {
_demux.on("error", _err => {
reject();
});
@ -300,14 +296,14 @@ test("demux() should emit one drain event after slowProcessorSpeed * highWaterMa
for (const item of input) {
const res = _demux.write(item);
if (!res) {
await new Promise((resolv, rej) => {
await new Promise((resolv, _rej) => {
// This event should be received after all items in demux are processed
_demux.once("drain", () => {
expect(performance.now() - start).to.be.greaterThan(
slowProcessorSpeed * highWaterMark,
);
t.pass();
resolv();
resolv(null);
});
});
}
@ -332,7 +328,7 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
{ key: "a", mapped: [] },
];
let pendingReads = input.length;
const construct = (destKey: string) => {
const construct = (_destKey: string) => {
const first = map(
async (chunk: Chunk) => {
await sleep(50);
@ -354,7 +350,7 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
highWaterMark: 5,
});
_demux.on("error", err => {
_demux.on("error", _err => {
reject();
});
@ -364,7 +360,7 @@ test("demux() should emit one drain event when writing 6 items with highWaterMar
if (!res) {
await new Promise(_resolve => {
_demux.once("drain", () => {
_resolve();
_resolve(null);
expect(_demux._writableState.length).to.be.equal(0);
t.pass();
});
@ -683,13 +679,7 @@ test.cb("Demux should send data events", t => {
});
test.cb("demux() `finish` and `end` propagates", t => {
interface Chunk {
key: string;
mapped: number[];
}
t.plan(9);
t.plan(10);
const construct = (destKey: string) => {
const dest = map((chunk: any) => {
chunk.mapped.push(destKey);
@ -728,10 +718,10 @@ test.cb("demux() `finish` and `end` propagates", t => {
});
_demux.on("end", () => {
t.pass();
t.end();
});
sink.on("finish", () => {
t.pass();
t.end();
});
const input = [

2343
yarn.lock
View File

File diff suppressed because it is too large Load Diff