Compare commits

...

6 Commits

Author SHA1 Message Date
60f9e541a5 3.0.30 2023-11-14 10:51:23 +01:00
96ea67e135 fix(core): update 2023-11-14 10:51:23 +01:00
ba0a2023ad 3.0.29 2023-11-14 10:43:18 +01:00
a09c359847 fix(core): update 2023-11-14 10:43:17 +01:00
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", "name": "@push.rocks/smartstream",
"version": "3.0.27", "version": "3.0.30",
"private": false, "private": false,
"description": "simplifies access to node streams", "description": "simplifies access to node streams",
"main": "dist_ts/index.js", "main": "dist_ts/index.js",

View File

@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/smartstream', name: '@push.rocks/smartstream',
version: '3.0.27', version: '3.0.30',
description: 'simplifies access to node streams' 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>>(); private asyncWritePromiseObjectmap = new plugins.lik.ObjectMap<Promise<any>>();
// Ensure the _write method types the chunk as TInput and encodes TOutput // Ensure the _write method types the chunk as TInput and encodes TOutput
public async _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) { 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(); callback();
}, },
push: async (pushArg: TOutput) => { push: async (pushArg: TOutput) => {
const canPushMore = this.backpressuredArray.push(pushArg); await this.backpressuredPush(pushArg);
if (!canPushMore) {
this.debugLog(`${this.options.name}: cannot push more`);
await this.backpressuredArray.waitForSpace();
this.debugLog(`${this.options.name}: can push more again`);
} }
},
}; };
try { try {