fix(core): update

This commit is contained in:
Philipp Kunz 2023-11-13 19:06:02 +01:00
parent bc09033af0
commit 2e1aa4a8ff
5 changed files with 7 additions and 46 deletions

View File

@ -10,41 +10,3 @@ hi+wow
hi+wow hi+wow
hi+wow hi+wow
hi+wow hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow
hi+wow

View File

@ -19,6 +19,7 @@ tap.test('should run backpressure test', async (toolsArg) => {
writeFunction: async (chunk, tools) => { writeFunction: async (chunk, tools) => {
await new Promise((resolve) => setTimeout(resolve, 20)); // Slow processing await new Promise((resolve) => setTimeout(resolve, 20)); // Slow processing
console.log(`processed chunk ${chunk} in stream 2`); console.log(`processed chunk ${chunk} in stream 2`);
// await tools.push(chunk);
return chunk; return chunk;
}, },
}); // This stream processes data more slowly }); // 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 () => { tap.test('should create a SmartStream from a Buffer', async () => {
const bufferData = Buffer.from('This is a test buffer'); const bufferData = Buffer.from('This is a test buffer');
const smartStream = SmartDuplex.fromBuffer(bufferData, { const smartStream = SmartDuplex.fromBuffer(bufferData, {});
handleBackpressure: false,
});
let receivedData = Buffer.alloc(0); let receivedData = Buffer.alloc(0);

View File

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

View File

@ -115,22 +115,22 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
const tools: IStreamTools = { const tools: IStreamTools = {
truncate: () => callback(), truncate: () => callback(),
push: async (pipeObject) => { push: async (pipeObject) => {
this.push(pipeObject); this.backpressuredArray.push(pipeObject);
}, },
}; };
try { try {
const finalChunk = await this.options.finalFunction(tools); const finalChunk = await this.options.finalFunction(tools);
if (finalChunk) { if (finalChunk) {
this.push(finalChunk); this.backpressuredArray.push(finalChunk);
} }
} catch (err) { } catch (err) {
this.push(null); this.backpressuredArray.push(null);
callback(err); callback(err);
return; return;
} }
} }
this.push(null); this.backpressuredArray.push(null);
callback(); callback();
} }
} }