Compare commits

...

2 Commits

Author SHA1 Message Date
7a14e67f4f 3.0.17 2023-11-11 20:44:01 +01:00
465ccfec40 fix(core): update 2023-11-11 20:44:00 +01:00
3 changed files with 6 additions and 2 deletions

View File

@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartstream", "name": "@push.rocks/smartstream",
"version": "3.0.16", "version": "3.0.17",
"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.16', version: '3.0.17',
description: 'simplifies access to node streams' description: 'simplifies access to node streams'
} }

View File

@ -129,12 +129,15 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
} }
} }
public notBackpressured = true;
public push(chunkArg?: TOutput | null): boolean { public push(chunkArg?: TOutput | null): boolean {
const result = super.push(chunkArg); const result = super.push(chunkArg);
if (!result && this.handleBackpressure) { if (!result && this.handleBackpressure) {
this.notBackpressured = false;
this.pause(); this.pause();
// Listen for 'drain' event to resume // Listen for 'drain' event to resume
this.once('drain', () => { this.once('drain', () => {
this.notBackpressured = true;
this.resume(); // Resume the source of data this.resume(); // Resume the source of data
}); });
} }
@ -164,6 +167,7 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
} catch (err) { } catch (err) {
callback(err); callback(err);
} }
return this.notBackpressured;
} }
public async _final(callback: (error?: Error | null) => void) { public async _final(callback: (error?: Error | null) => void) {