Compare commits
2 Commits
Author | SHA1 | Date | |
---|---|---|---|
df0ddf04b3 | |||
2e1aa4a8ff |
@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "@push.rocks/smartstream",
|
"name": "@push.rocks/smartstream",
|
||||||
"version": "3.0.23",
|
"version": "3.0.24",
|
||||||
"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",
|
||||||
|
@ -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
|
|
||||||
|
@ -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
|
||||||
|
@ -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);
|
||||||
|
|
||||||
|
@ -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'
|
||||||
}
|
}
|
||||||
|
@ -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();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Reference in New Issue
Block a user