Compare commits

...

2 Commits

Author SHA1 Message Date
e2b4d772b3 3.0.28 2023-11-14 10:29:44 +01:00
0f46b62b2d fix(core): update 2023-11-14 10:29:44 +01:00
3 changed files with 13 additions and 9 deletions

View File

@ -1,6 +1,6 @@
{
"name": "@push.rocks/smartstream",
"version": "3.0.27",
"version": "3.0.28",
"private": false,
"description": "simplifies access to node streams",
"main": "dist_ts/index.js",

View File

@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@push.rocks/smartstream',
version: '3.0.27',
version: '3.0.28',
description: 'simplifies access to node streams'
}

View File

@ -66,6 +66,15 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
}
}
public async backpressuredPush (pushArg: TOutput) {
const canPushMore = this.backpressuredArray.push(pushArg);
if (!canPushMore) {
this.debugLog(`${this.options.name}: cannot push more`);
await this.backpressuredArray.waitForSpace();
this.debugLog(`${this.options.name}: can push more again`);
}
};
private asyncWritePromiseObjectmap = new plugins.lik.ObjectMap<Promise<any>>();
// Ensure the _write method types the chunk as TInput and encodes TOutput
public async _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) {
@ -81,13 +90,8 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
callback();
},
push: async (pushArg: TOutput) => {
const canPushMore = this.backpressuredArray.push(pushArg);
if (!canPushMore) {
this.debugLog(`${this.options.name}: cannot push more`);
await this.backpressuredArray.waitForSpace();
this.debugLog(`${this.options.name}: can push more again`);
}
},
await this.backpressuredPush(pushArg);
}
};
try {