3.0.10

package.json

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

test/test.streamfunction.ts

@@ -16,7 +16,7 @@ tap.test('should handle a read stream', async (tools) => {
         streamTools.push('wow =========== \n');
         return Buffer.from(result);
       },
-      streamEndFunction: async (tools) => {
+      finalFunction: async (tools) => {
         return Buffer.from('this is the end');
       },
     }),
@@ -24,7 +24,7 @@ tap.test('should handle a read stream', async (tools) => {
       writeAndTransformFunction: async (chunkStringArg) => {
         console.log(chunkStringArg.toString());
       },
-      streamEndFunction: async (tools) => {
+      finalFunction: async (tools) => {
         tools.push(null);
       },
     })

ts/00_commitinfo_data.ts

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

ts/index.ts

@@ -2,3 +2,5 @@ export * from './smartstream.classes.passthrough.js';
 export * from './smartstream.classes.smartduplex.js';
 export * from './smartstream.classes.streamwrapper.js';
 export * from './smartstream.classes.streamintake.js';
+
+export * from './smartstream.functions.js'

ts/smartstream.classes.passthrough.ts

@@ -13,7 +13,9 @@ export class PassThrough extends plugins.stream.Duplex {
     if (this.push(chunk, encoding)) {
       callback();
     } else {
-      this.once('drain', callback);
+      this.once('drain', () => {
+        callback();
+      });
     }
   }
 }

ts/smartstream.classes.smartduplex.ts

@@ -17,7 +17,7 @@ export interface IStreamEndFunction<rT> {
 export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions {
   readFunction?: () => Promise<void>;
   writeAndTransformFunction?: IWriteAndTransformFunction<TInput, TOutput>;
-  streamEndFunction?: IStreamEndFunction<TOutput>;
+  finalFunction?: IStreamEndFunction<TOutput>;
   // Add other custom options if necessary
 }
 
@@ -117,7 +117,7 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
     super(optionsArg);
     this.readFunction = optionsArg?.readFunction;
     this.writeAndTransformFunction = optionsArg?.writeAndTransformFunction;
-    this.streamEndFunction = optionsArg?.streamEndFunction;
+    this.streamEndFunction = optionsArg?.finalFunction;
   }
 
   public async _read(size: number): Promise<void> {
@@ -170,8 +170,8 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
         callback(err);
       }
     } else {
-      this.push(null),
       callback();
     }
+    this.push(null);
   }
 }

ts/smartstream.functions.ts

@@ -0,0 +1,26 @@
+import { Transform, type TransformCallback, type TransformOptions } from 'stream';
+
+export interface AsyncTransformFunction<TInput, TOutput> {
+  (chunkArg: TInput): Promise<TOutput>;
+}
+
+export function createTransformFunction<TInput, TOutput>(
+  asyncFunction: AsyncTransformFunction<TInput, TOutput>,
+  options?: TransformOptions
+): Transform {
+  const transformStream = new Transform({
+    ...options,
+    objectMode: true, // Ensure we operate in object mode
+    async transform(chunk: TInput, encoding: string, callback: TransformCallback) {
+      try {
+        const transformed = await asyncFunction(chunk);
+        this.push(transformed);
+        callback();
+      } catch (error) {
+        callback(error instanceof Error ? error : new Error(String(error)));
+      }
+    }
+  });
+
+  return transformStream;
+}