Compare commits

...

4 Commits

Author SHA1 Message Date
0ceeacd5a0 3.0.31 2024-02-25 20:14:20 +01:00
287695e445 fix(core): update 2024-02-25 20:14:19 +01:00
60f9e541a5 3.0.30 2023-11-14 10:51:23 +01:00
96ea67e135 fix(core): update 2023-11-14 10:51:23 +01:00
5 changed files with 442 additions and 394 deletions

View File

@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartstream", "name": "@push.rocks/smartstream",
"version": "3.0.29", "version": "3.0.31",
"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",
@ -22,15 +22,15 @@
}, },
"homepage": "https://gitlab.com/push.rocks/smartstream#readme", "homepage": "https://gitlab.com/push.rocks/smartstream#readme",
"devDependencies": { "devDependencies": {
"@git.zone/tsbuild": "^2.1.66", "@git.zone/tsbuild": "^2.1.72",
"@git.zone/tsrun": "^1.2.44", "@git.zone/tsrun": "^1.2.44",
"@git.zone/tstest": "^1.0.84", "@git.zone/tstest": "^1.0.86",
"@push.rocks/smartfile": "^11.0.0", "@push.rocks/smartfile": "^11.0.4",
"@push.rocks/tapbundle": "^5.0.15", "@push.rocks/tapbundle": "^5.0.15",
"@types/node": "^20.9.0" "@types/node": "^20.11.20"
}, },
"dependencies": { "dependencies": {
"@push.rocks/lik": "^6.0.12", "@push.rocks/lik": "^6.0.14",
"@push.rocks/smartpromise": "^4.0.3", "@push.rocks/smartpromise": "^4.0.3",
"@push.rocks/smartrx": "^3.0.7" "@push.rocks/smartrx": "^3.0.7"
}, },

798
pnpm-lock.yaml generated

File diff suppressed because it is too large Load Diff

View File

@ -20,7 +20,7 @@ tap.test('should run backpressure test', async (toolsArg) => {
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); await tools.push(chunk);
// return chunk; // return chunk, optionally return ;
}, },
}); // This stream processes data more slowly }); // This stream processes data more slowly
const stream3 = new SmartDuplex({ const stream3 = new SmartDuplex({

View File

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

View File

@ -15,10 +15,28 @@ export interface IStreamFinalFunction<rT> {
} }
export interface ISmartDuplexOptions<TInput, TOutput> extends DuplexOptions { export interface ISmartDuplexOptions<TInput, TOutput> extends DuplexOptions {
/**
* wether to print debug logs
*/
debug?: boolean; debug?: boolean;
/**
* the name of the stream
*/
name?: string; name?: string;
/**
* a function that is being called to read more stuff from whereever to be processed by the stream
* @returns
*/
readFunction?: () => Promise<void>; readFunction?: () => Promise<void>;
/**
* the write function is called for every chunk that is being written to the stream
* it can push or return chunks (but does not have to) to be written to the readable side of the stream
*/
writeFunction?: IStreamWriteFunction<TInput, TOutput>; writeFunction?: IStreamWriteFunction<TInput, TOutput>;
/**
* a final function that is being sent to the end of the stream
*/
finalFunction?: IStreamFinalFunction<TOutput>; finalFunction?: IStreamFinalFunction<TOutput>;
// Add other custom options if necessary // Add other custom options if necessary
} }
@ -46,10 +64,10 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
constructor(optionsArg?: ISmartDuplexOptions<TInput, TOutput>) { constructor(optionsArg?: ISmartDuplexOptions<TInput, TOutput>) {
super(Object.assign({ super(Object.assign({
highWaterMark: 4, highWaterMark: 1,
}, optionsArg)); }, optionsArg));
this.options = optionsArg; this.options = optionsArg;
this.backpressuredArray = new plugins.lik.BackpressuredArray<TOutput>(this.options.highWaterMark || 4) this.backpressuredArray = new plugins.lik.BackpressuredArray<TOutput>(this.options.highWaterMark || 1)
} }
public async _read(size: number): Promise<void> { public async _read(size: number): Promise<void> {