push.rocks/smartstream
2
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

fix(core): update

Browse Source
This commit is contained in:
Philipp Kunz 2023-11-14 10:29:44 +01:00
parent 9bf37469c6
commit 0f46b62b2d
2 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ts/00_commitinfo_data.ts
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.28',
description: 'simplifies access to node streams' description: 'simplifies access to node streams'
} }

18
ts/smartstream.classes.smartduplex.ts
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 {