fix(core): update

ts/smartstream.classes.smartduplex.ts

@@ -0,0 +1,177 @@
+import * as plugins from './smartstream.plugins.js';
+import { Duplex, type DuplexOptions } from 'stream';
+
+export interface IStreamTools {
+  truncate: () => void;
+  push: (pipeObject: any) => void;
+}
+
+export interface IWriteAndTransformFunction<T, rT> {
+  (chunkArg: T, toolsArg: IStreamTools): Promise<rT>;
+}
+
+export interface IStreamEndFunction<rT> {
+  (toolsArg: IStreamTools): Promise<rT>;
+}
+
+export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions {
+  readFunction?: () => Promise<void>;
+  writeAndTransformFunction?: IWriteAndTransformFunction<TInput, TOutput>;
+  streamEndFunction?: IStreamEndFunction<TOutput>;
+  // Add other custom options if necessary
+}
+
+export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
+  // STATIC
+  static fromBuffer(buffer: Buffer, options?: DuplexOptions): SmartDuplex {
+    const smartStream = new SmartDuplex(options);
+    process.nextTick(() => {
+      smartStream.push(buffer);
+      smartStream.push(null); // Signal the end of the data
+    });
+    return smartStream;
+  }
+
+  static fromObservable(
+    observable: plugins.smartrx.rxjs.Observable<any>,
+    options?: DuplexOptions
+  ): SmartDuplex {
+    const smartStream = new SmartDuplex(options);
+    smartStream.observableSubscription = observable.subscribe({
+      next: (data) => {
+        if (!smartStream.push(data)) {
+          // Pause the observable if the stream buffer is full
+          smartStream.observableSubscription?.unsubscribe();
+          smartStream.once('drain', () => {
+            // Resume the observable when the stream buffer is drained
+            smartStream.observableSubscription?.unsubscribe();
+            smartStream.observableSubscription = observable.subscribe((data) => {
+              smartStream.push(data);
+            });
+          });
+        }
+      },
+      error: (err) => {
+        smartStream.emit('error', err);
+      },
+      complete: () => {
+        smartStream.push(null); // Signal the end of the data
+      },
+    });
+
+    return smartStream;
+  }
+
+  static fromReplaySubject(
+    replaySubject: plugins.smartrx.rxjs.ReplaySubject<any>,
+    options?: DuplexOptions
+  ): SmartDuplex {
+    const smartStream = new SmartDuplex(options);
+    let isBackpressured = false;
+
+    // Subscribe to the ReplaySubject
+    const subscription = replaySubject.subscribe({
+      next: (data) => {
+        const canPush = smartStream.push(data);
+        if (!canPush) {
+          // If push returns false, pause the subscription because of backpressure
+          isBackpressured = true;
+          subscription.unsubscribe();
+        }
+      },
+      error: (err) => {
+        smartStream.emit('error', err);
+      },
+      complete: () => {
+        smartStream.push(null); // End the stream when the ReplaySubject completes
+      },
+    });
+
+    // Listen for 'drain' event to resume the subscription if it was paused
+    smartStream.on('drain', () => {
+      if (isBackpressured) {
+        isBackpressured = false;
+        // Resubscribe to the ReplaySubject since we previously paused
+        smartStream.observableSubscription = replaySubject.subscribe({
+          next: (data) => {
+            if (!smartStream.push(data)) {
+              smartStream.observableSubscription?.unsubscribe();
+              isBackpressured = true;
+            }
+          },
+          // No need to repeat error and complete handling here because it's already set up above
+        });
+      }
+    });
+
+    return smartStream;
+  }
+
+  // INSTANCE
+  private readFunction?: () => Promise<void>;
+  private writeAndTransformFunction?: IWriteAndTransformFunction<TInput, TOutput>;
+  private streamEndFunction?: IStreamEndFunction<TOutput>;
+  private observableSubscription?: plugins.smartrx.rxjs.Subscription;
+
+  constructor(optionsArg?: SmartStreamOptions<TInput, TOutput>) {
+    super(optionsArg);
+    this.readFunction = optionsArg?.readFunction;
+    this.writeAndTransformFunction = optionsArg?.writeAndTransformFunction;
+    this.streamEndFunction = optionsArg?.streamEndFunction;
+  }
+
+  public async _read(size: number): Promise<void> {
+    if (this.readFunction) {
+      await this.readFunction();
+    }
+  }
+
+  // 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.writeAndTransformFunction) {
+      return callback(new Error('No stream function provided'));
+    }
+
+    const tools: IStreamTools = {
+      truncate: () => {
+        this.push(null);
+        callback();
+      },
+      push: (pushArg: TOutput) => this.push(pushArg),
+    };
+
+    try {
+      const modifiedChunk = await this.writeAndTransformFunction(chunk, tools);
+      if (modifiedChunk) {
+        if (!this.push(modifiedChunk)) {
+          // Handle backpressure if necessary
+        }
+      }
+      callback();
+    } catch (err) {
+      callback(err);
+    }
+  }
+
+  public async _final(callback: (error?: Error | null) => void) {
+    if (this.streamEndFunction) {
+      const tools: IStreamTools = {
+        truncate: () => callback(),
+        push: (pipeObject) => this.push(pipeObject),
+      };
+
+      try {
+        const finalChunk = await this.streamEndFunction(tools);
+        if (finalChunk) {
+          this.push(finalChunk);
+        }
+        callback();
+      } catch (err) {
+        callback(err);
+      }
+    } else {
+      this.push(null),
+      callback();
+    }
+  }
+}