push.rocks/smartstream
2
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

fix(core): update

Browse Source
This commit is contained in:
Philipp Kunz 2023-11-07 21:46:46 +01:00
parent d238662bea
commit 49d235411f
3 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
test/test.streamfunction.ts
View File

@ -10,7 +10,7 @@ tap.test('should handle a read stream', async (tools) => {
const streamWrapper = new smartstream.StreamWrapper([ const streamWrapper = new smartstream.StreamWrapper([
smartfile.fsStream.createReadStream('./test/assets/readabletext.txt'), smartfile.fsStream.createReadStream('./test/assets/readabletext.txt'),
new smartstream.SmartDuplex({ new smartstream.SmartDuplex({
writeAndTransformFunction: async (chunkStringArg: Buffer, streamTools) => { writeFunction: async (chunkStringArg: Buffer, streamTools) => {
// do something with the stream here // do something with the stream here
const result = chunkStringArg.toString().substr(0, 100); const result = chunkStringArg.toString().substr(0, 100);
streamTools.push('wow =========== \n'); streamTools.push('wow =========== \n');
@ -21,7 +21,7 @@ tap.test('should handle a read stream', async (tools) => {
}, },
}), }),
new smartstream.SmartDuplex({ new smartstream.SmartDuplex({
writeAndTransformFunction: async (chunkStringArg) => { writeFunction: async (chunkStringArg) => {
console.log(chunkStringArg.toString()); console.log(chunkStringArg.toString());
}, },
finalFunction: async (tools) => { finalFunction: async (tools) => {
@ -37,7 +37,7 @@ tap.test('should create a valid Intake', async (tools) => {
testIntake.pipe( testIntake.pipe(
new smartstream.SmartDuplex({ new smartstream.SmartDuplex({
objectMode: true, objectMode: true,
writeAndTransformFunction: async (chunkStringArg: string, streamTools) => { writeFunction: async (chunkStringArg: string, streamTools) => {
await tools.delayFor(100); await tools.delayFor(100);
console.log(chunkStringArg); console.log(chunkStringArg);
return chunkStringArg; return chunkStringArg;

2
ts/00_commitinfo_data.ts
View File

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

14
ts/smartstream.classes.smartduplex.ts
View File

@ -6,18 +6,18 @@ export interface IStreamTools {
push: (pipeObject: any) => void; push: (pipeObject: any) => void;
} }
export interface IWriteAndTransformFunction<T, rT> { export interface IStreamWriteFunction<T, rT> {
(chunkArg: T, toolsArg: IStreamTools): Promise<rT>; (chunkArg: T, toolsArg: IStreamTools): Promise<rT>;
} }
export interface IStreamEndFunction<rT> { export interface IStreamFinalFunction<rT> {
(toolsArg: IStreamTools): Promise<rT>; (toolsArg: IStreamTools): Promise<rT>;
} }
export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions { export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions {
readFunction?: () => Promise<void>; readFunction?: () => Promise<void>;
writeAndTransformFunction?: IWriteAndTransformFunction<TInput, TOutput>; writeFunction?: IStreamWriteFunction<TInput, TOutput>;
finalFunction?: IStreamEndFunction<TOutput>; finalFunction?: IStreamFinalFunction<TOutput>;
// Add other custom options if necessary // Add other custom options if necessary
} }
@ -109,14 +109,14 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
// INSTANCE // INSTANCE
private readFunction?: () => Promise<void>; private readFunction?: () => Promise<void>;
private writeFunction?: IWriteAndTransformFunction<TInput, TOutput>; private writeFunction?: IStreamWriteFunction<TInput, TOutput>;
private finalFunction?: IStreamEndFunction<TOutput>; private finalFunction?: IStreamFinalFunction<TOutput>;
private observableSubscription?: plugins.smartrx.rxjs.Subscription; private observableSubscription?: plugins.smartrx.rxjs.Subscription;
constructor(optionsArg?: SmartStreamOptions<TInput, TOutput>) { constructor(optionsArg?: SmartStreamOptions<TInput, TOutput>) {
super(optionsArg); super(optionsArg);
this.readFunction = optionsArg?.readFunction; this.readFunction = optionsArg?.readFunction;
this.writeFunction = optionsArg?.writeAndTransformFunction; this.writeFunction = optionsArg?.writeFunction;
this.finalFunction = optionsArg?.finalFunction; this.finalFunction = optionsArg?.finalFunction;
} }