3.0.18

package.json

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

test/test.smartstream.ts

@@ -5,7 +5,9 @@ import * as fs from 'fs';
 
 tap.test('should create a SmartStream from a Buffer', async () => {
   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);
   
@@ -25,7 +27,9 @@ tap.test('should create a SmartStream from an Observable', async () => {
   const observableData = 'Observable test data';
   const testObservable = smartrx.rxjs.of(Buffer.from(observableData));
   
-  const smartStream = SmartDuplex.fromObservable(testObservable);
+  const smartStream = SmartDuplex.fromObservable(testObservable, {
+    handleBackpressure: false,
+  });
   
   let receivedData = Buffer.alloc(0);
 

test/test.streamfunction.ts

@@ -29,7 +29,7 @@ tap.test('should handle a read stream', async (tools) => {
       },
     })
   ]);
-  // await streamWrapper.run();
+  await streamWrapper.run();
 });
 
 tap.test('should create a valid Intake', async (tools) => {

ts/00_commitinfo_data.ts

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

ts/smartstream.classes.smartduplex.ts

@@ -14,7 +14,7 @@ export interface IStreamFinalFunction<rT> {
   (toolsArg: IStreamTools): Promise<rT>;
 }
 
-export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions {
+export interface ISmartDuplexOptions<TInput, TOutput> extends DuplexOptions {
   handleBackpressure?: boolean;
   readFunction?: () => Promise<void>;
   writeFunction?: IStreamWriteFunction<TInput, TOutput>;
@@ -24,23 +24,23 @@ export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions {
 
 export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
   // STATIC
-  static fromBuffer(buffer: Buffer, options?: DuplexOptions): SmartDuplex {
+  static fromBuffer(buffer: Buffer, options?: ISmartDuplexOptions<any, any>): SmartDuplex {
-    const smartStream = new SmartDuplex(options);
+    const smartDuplex = new SmartDuplex(options);
     process.nextTick(() => {
-      smartStream.push(buffer);
+      smartDuplex.push(buffer);
-      smartStream.push(null); // Signal the end of the data
+      smartDuplex.push(null); // Signal the end of the data
     });
-    return smartStream;
+    return smartDuplex;
   }
 
   static fromObservable(
     observable: plugins.smartrx.rxjs.Observable<any>,
-    options?: DuplexOptions
+    options?: ISmartDuplexOptions<any, any>
   ): SmartDuplex {
     const smartStream = new SmartDuplex(options);
     smartStream.observableSubscription = observable.subscribe({
       next: (data) => {
-        if (!smartStream.push(data)) {
+        if (!smartStream.push(data) && smartStream.handleBackpressure) {
           // Pause the observable if the stream buffer is full
           smartStream.observableSubscription?.unsubscribe();
           smartStream.once('drain', () => {
@@ -115,7 +115,7 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
   private finalFunction?: IStreamFinalFunction<TOutput>;
   private observableSubscription?: plugins.smartrx.rxjs.Subscription;
 
-  constructor(optionsArg?: SmartStreamOptions<TInput, TOutput>) {
+  constructor(optionsArg?: ISmartDuplexOptions<TInput, TOutput>) {
     super(optionsArg);
     this.readFunction = optionsArg?.readFunction;
     this.writeFunction = optionsArg?.writeFunction;
@@ -129,6 +129,24 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
     }
   }
 
+  public notBackpressured = true;
+  public get backpressured(): boolean {
+    return !this.notBackpressured;
+  }
+  public push(chunkArg?: TOutput | null): boolean {
+    const result = super.push(chunkArg);
+    if (!result && this.handleBackpressure) {
+      this.notBackpressured = false;
+      this.pause();
+      // Listen for 'drain' event to resume
+      this.once('drain', () => {
+        this.notBackpressured = true;
+        this.resume(); // Resume the source of data
+      });
+    }
+    return result;
+  }
+
   // Ensure the _write method types the chunk as TInput and encodes TOutput
   public async _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) {
     if (!this.writeFunction) {
@@ -146,18 +164,21 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
     try {
       const modifiedChunk = await this.writeFunction(chunk, tools);
       if (modifiedChunk) {
-        if (!this.push(modifiedChunk) && this.handleBackpressure) {
+        this.push(modifiedChunk)
-          this.pause();
+        if (this.backpressured && this.handleBackpressure) {
-          // Listen for 'drain' event to resume
           this.once('drain', () => {
-            this.resume(); // Resume the source of data
+            callback();
           });
+        } else {
+          callback();
         }      
+      } else {
+        callback();
       }
-      callback();
     } catch (err) {
       callback(err);
     }
+    return this.notBackpressured;
   }
 
   public async _final(callback: (error?: Error | null) => void) {

ts/smartstream.functions.ts

@@ -1,4 +1,5 @@
 import { Transform, type TransformCallback, type TransformOptions } from 'stream';
+import { SmartDuplex } from './smartstream.classes.smartduplex.js';
 
 export interface AsyncTransformFunction<TInput, TOutput> {
   (chunkArg: TInput): Promise<TOutput>;
@@ -7,20 +8,14 @@ export interface AsyncTransformFunction<TInput, TOutput> {
 export function createTransformFunction<TInput, TOutput>(
   asyncFunction: AsyncTransformFunction<TInput, TOutput>,
   options?: TransformOptions
-): Transform {
+): SmartDuplex {
-  const transformStream = new Transform({
+  const smartDuplexStream = new SmartDuplex({
     ...options,
-    objectMode: true, // Ensure we operate in object mode
+    writeFunction: async (chunkArg, toolsArg) => {
-    async transform(chunk: TInput, encoding: string, callback: TransformCallback) {
+      const result = await asyncFunction(chunkArg);
-      try {
+      return result;
-        const transformed = await asyncFunction(chunk);
-        this.push(transformed);
-        callback();
-      } catch (error) {
-        callback(error instanceof Error ? error : new Error(String(error)));
-      }
     }
   });
 
-  return transformStream;
+  return smartDuplexStream;
 }