Compare commits

...

4 Commits

Author SHA1 Message Date
c77e0f2ba6 3.0.25 2023-11-13 19:12:24 +01:00
196fb6d396 fix(core): update 2023-11-13 19:12:23 +01:00
df0ddf04b3 3.0.24 2023-11-13 19:06:02 +01:00
2e1aa4a8ff fix(core): update 2023-11-13 19:06:02 +01:00
5 changed files with 9 additions and 12 deletions

View File

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

View File

@ -19,6 +19,7 @@ tap.test('should run backpressure test', async (toolsArg) => {
writeFunction: async (chunk, tools) => {
await new Promise((resolve) => setTimeout(resolve, 20)); // Slow processing
console.log(`processed chunk ${chunk} in stream 2`);
// await tools.push(chunk);
return chunk;
},
}); // This stream processes data more slowly

View File

@ -5,9 +5,7 @@ import * as fs from 'fs';
tap.test('should create a SmartStream from a Buffer', async () => {
const bufferData = Buffer.from('This is a test buffer');
const smartStream = SmartDuplex.fromBuffer(bufferData, {
handleBackpressure: false,
});
const smartStream = SmartDuplex.fromBuffer(bufferData, {});
let receivedData = Buffer.alloc(0);

View File

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

View File

@ -59,9 +59,7 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
let canPushMore = true;
while(this.backpressuredArray.data.length > 0 && canPushMore) {
const nextChunk = this.backpressuredArray.shift();
if (nextChunk) {
canPushMore = this.push(nextChunk);
}
canPushMore = this.push(nextChunk);
}
}
@ -115,22 +113,22 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
const tools: IStreamTools = {
truncate: () => callback(),
push: async (pipeObject) => {
this.push(pipeObject);
this.backpressuredArray.push(pipeObject);
},
};
try {
const finalChunk = await this.options.finalFunction(tools);
if (finalChunk) {
this.push(finalChunk);
this.backpressuredArray.push(finalChunk);
}
} catch (err) {
this.push(null);
this.backpressuredArray.push(null);
callback(err);
return;
}
}
this.push(null);
this.backpressuredArray.push(null);
callback();
}
}