Compare commits
2 Commits
Author | SHA1 | Date | |
---|---|---|---|
e2b4d772b3 | |||
0f46b62b2d |
@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "@push.rocks/smartstream",
|
"name": "@push.rocks/smartstream",
|
||||||
"version": "3.0.27",
|
"version": "3.0.28",
|
||||||
"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",
|
||||||
|
@ -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'
|
||||||
}
|
}
|
||||||
|
@ -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 {
|
||||||
|
Reference in New Issue
Block a user