3.2.0

changelog.md

@@ -1,5 +1,31 @@
 # Changelog
 
+## 2024-10-16 - 3.2.0 - feat(SmartDuplex)
+Added method to create SmartDuplex from a WebReadableStream.
+
+- Implemented a static method in SmartDuplex to allow creating an instance from a WebReadableStream.
+- This addition enhances the capability of SmartDuplex to integrate with web streams, facilitating seamless stream manipulation across environments.
+
+## 2024-10-14 - 3.1.2 - fix(WebDuplexStream)
+Fix variable naming inconsistency in WebDuplexStream test
+
+- Changed variable names from 'transformStream' to 'webDuplexStream' for consistency.
+- Renamed 'writableStream' and 'readableStream' to 'writer' and 'reader' respectively.
+
+## 2024-10-13 - 3.1.1 - fix(WebDuplexStream)
+Improved read/write interface and error handling in WebDuplexStream
+
+- Enhanced the IStreamToolsRead and IStreamToolsWrite interfaces for better Promise handling
+- Refined readFunction and writeFunction handling to accommodate asynchronous data processing and error propagation
+- Added internal _startReading method to facilitate initial data handling if readFunction is present
+- Maintained backward compatibility while ensuring data continuity when no writeFunction is specified
+
+## 2024-10-13 - 3.1.0 - feat(core)
+Add support for creating Web ReadableStream from a file
+
+- Introduced a new helper function `createWebReadableStreamFromFile` that allows for creating a Web ReadableStream from a file path.
+- Updated exports in `ts/index.ts` to include `nodewebhelpers` which provides the new web stream feature.
+
 ## 2024-10-13 - 3.0.46 - fix(WebDuplexStream)
 Fix errors in WebDuplexStream transformation and test logic
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartstream",
-  "version": "3.0.46",
+  "version": "3.2.0",
   "private": false,
   "description": "A library to simplify the creation and manipulation of Node.js streams, providing utilities for handling transform, duplex, and readable/writable streams effectively in TypeScript.",
   "type": "module",

test/test.ts_web.both.ts

@@ -28,22 +28,22 @@ tap.test('WebDuplexStream should handle transform with a write function', async
   const input = [1, 2, 3, 4, 5];
   const expectedOutput = [2, 4, 6, 8, 10];
 
-  const transformStream = new webstream.WebDuplexStream<number, number>({
+  const webDuplexStream = new webstream.WebDuplexStream<number, number>({
     writeFunction: async (chunk, { push }) => {
       // Push the doubled number into the stream
       push(chunk * 2);
     },
   });
 
-  const writableStream = transformStream.writable.getWriter();
+  const writer = webDuplexStream.writable.getWriter();
-  const readableStream = transformStream.readable.getReader();
+  const reader = webDuplexStream.readable.getReader();
 
   const output: number[] = [];
 
   // Read from the stream asynchronously
   const readPromise = (async () => {
     while (true) {
-      const { value, done } = await readableStream.read();
+      const { value, done } = await reader.read();
       if (done) break;
       if (value !== undefined) {
         output.push(value);
@@ -53,9 +53,9 @@ tap.test('WebDuplexStream should handle transform with a write function', async
 
   // Write to the stream
   for (const num of input) {
-    await writableStream.write(num);
+    await writer.write(num);
   }
-  await writableStream.close();
+  await writer.close();
 
   // Wait for the reading to complete
   await readPromise;

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartstream',
-  version: '3.0.46',
+  version: '3.2.0',
   description: 'A library to simplify the creation and manipulation of Node.js streams, providing utilities for handling transform, duplex, and readable/writable streams effectively in TypeScript.'
 }

ts/index.ts

@@ -11,3 +11,7 @@ export * from './smartstream.functions.js';
 
 import * as plugins from './smartstream.plugins.js';
 export const webstream = plugins.webstream;
+import * as nodewebhelpers from './smartstream.nodewebhelpers.js';
+export {
+  nodewebhelpers,
+}

ts/smartstream.classes.smartduplex.ts

@@ -52,6 +52,31 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
     return smartDuplex;
   }
 
+  public static fromWebReadableStream<T = any>(
+    readableStream: ReadableStream<T>
+  ): SmartDuplex<T, T> {
+    const smartDuplex = new SmartDuplex<T, T>({
+      readFunction: async () => {
+        const reader = readableStream.getReader();
+        try {
+          while (true) {
+            const { value, done } = await reader.read();
+            if (value !== undefined) {
+              smartDuplex.push(value);
+            }
+            if (done) {
+              smartDuplex.end();
+              break;
+            }
+          }
+        } finally {
+          reader.releaseLock();
+        }
+      },
+    });
+    return smartDuplex;
+  }
+
   // INSTANCE
   private backpressuredArray: plugins.lik.BackpressuredArray<TOutput>; // an array that only takes a defined amount of items
   public options: ISmartDuplexOptions<TInput, TOutput>;

ts/smartstream.functions.ts

@@ -1,4 +1,4 @@
-import { Transform, type TransformCallback, type TransformOptions } from 'stream';
+import { type TransformOptions } from 'stream';
 import { SmartDuplex } from './smartstream.classes.smartduplex.js';
 
 export interface AsyncTransformFunction<TInput, TOutput> {

ts/smartstream.nodewebhelpers.ts

@@ -0,0 +1,34 @@
+import { createReadStream } from 'fs';
+
+/**
+ * Creates a Web ReadableStream from a file.
+ *
+ * @param filePath - The path to the file to be read
+ * @returns A Web ReadableStream that reads the file in chunks
+ */
+export function createWebReadableStreamFromFile(filePath: string): ReadableStream<Uint8Array> {
+  const fileStream = createReadStream(filePath);
+
+  return new ReadableStream({
+    start(controller) {
+      // When data is available, enqueue it into the Web ReadableStream
+      fileStream.on('data', (chunk) => {
+        controller.enqueue(chunk as Uint8Array);
+      });
+
+      // When the file stream ends, close the Web ReadableStream
+      fileStream.on('end', () => {
+        controller.close();
+      });
+
+      // If there's an error, error the Web ReadableStream
+      fileStream.on('error', (err) => {
+        controller.error(err);
+      });
+    },
+    cancel() {
+      // If the Web ReadableStream is canceled, destroy the file stream
+      fileStream.destroy();
+    }
+  });
+}

ts_web/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartstream',
-  version: '3.0.46',
+  version: '3.2.0',
   description: 'A library to simplify the creation and manipulation of Node.js streams, providing utilities for handling transform, duplex, and readable/writable streams effectively in TypeScript.'
 }

ts_web/classes.webduplexstream.ts

@@ -1,25 +1,22 @@
 import * as plugins from './plugins.js';
 
-
 // ========================================
-// READ
+// Interfaces for Read functionality
 // ========================================
 export interface IStreamToolsRead<TInput, TOutput> {
   done: () => void;
-  write: (writeArg: TInput) => void;
+  write: (writeArg: TInput) => Promise<void>;
 }
 
 /**
- * the read function is called anytime
+ * The read function is called when data needs to be read into the stream.
- *   -> the WebDuplexStream is being read from
- *      and at the same time if nothing is enqueued
  */
 export interface IStreamReadFunction<TInput, TOutput> {
   (toolsArg: IStreamToolsRead<TInput, TOutput>): Promise<void>;
 }
 
 // ========================================
-// WRITE
+// Interfaces for Write functionality
 // ========================================
 export interface IStreamToolsWrite<TInput, TOutput> {
   truncate: () => void;
@@ -27,15 +24,14 @@ export interface IStreamToolsWrite<TInput, TOutput> {
 }
 
 /**
- * the write function can return something.
+ * The write function is called whenever a chunk is written to the stream.
- * It is called anytime a chunk is written to the stream.
  */
 export interface IStreamWriteFunction<TInput, TOutput> {
   (chunkArg: TInput, toolsArg: IStreamToolsWrite<TInput, TOutput>): Promise<any>;
 }
 
 export interface IStreamFinalFunction<TInput, TOutput> {
-  (toolsArg: IStreamToolsWrite<TInput, TOutput>): Promise<TOutput>;
+  (toolsArg: IStreamToolsWrite<TInput, TOutput>): Promise<TOutput | void>;
 }
 
 export interface WebDuplexStreamOptions<TInput, TOutput> {
@@ -45,28 +41,15 @@ export interface WebDuplexStreamOptions<TInput, TOutput> {
 }
 
 export class WebDuplexStream<TInput = any, TOutput = any> extends TransformStream<TInput, TOutput> {
-  static fromUInt8Array(uint8Array: Uint8Array): WebDuplexStream<Uint8Array, Uint8Array> {
-    const stream = new WebDuplexStream<Uint8Array, Uint8Array>({
-      writeFunction: async (chunk, { push }) => {
-        push(chunk); // Directly push the chunk as is
-        return null;
-      }
-    });
-
-    const writer = stream.writable.getWriter();
-    writer.write(uint8Array).then(() => writer.close());
-
-    return stream;
-  }
-  
   // INSTANCE
   options: WebDuplexStreamOptions<TInput, TOutput>;
 
   constructor(optionsArg: WebDuplexStreamOptions<TInput, TOutput>) {
-    // here we call into the official web stream api
     super({
+      async start(controller) {
+        // Optionally initialize any state here
+      },
       async transform(chunk, controller) {
-        // Transformation logic remains unchanged
         if (optionsArg?.writeFunction) {
           const tools: IStreamToolsWrite<TInput, TOutput> = {
             truncate: () => controller.terminate(),
@@ -75,82 +58,78 @@ export class WebDuplexStream<TInput = any, TOutput = any> extends TransformStrea
 
           try {
             const writeReturnChunk = await optionsArg.writeFunction(chunk, tools);
-            if (writeReturnChunk) { // return chunk is optional
+            if (writeReturnChunk !== undefined && writeReturnChunk !== null) {
               controller.enqueue(writeReturnChunk);
             }
           } catch (err) {
             controller.error(err);
           }
         } else {
-          controller.error(new Error('No write function provided'));
+          // If no writeFunction is provided, pass the chunk through
+          controller.enqueue(chunk as unknown as TOutput);
         }
       },
       async flush(controller) {
-        // Flush logic remains unchanged
         if (optionsArg?.finalFunction) {
           const tools: IStreamToolsWrite<TInput, TOutput> = {
             truncate: () => controller.terminate(),
-            push: (pipeObject) => controller.enqueue(pipeObject),
+            push: (pushArg) => controller.enqueue(pushArg),
           };
 
-          optionsArg.finalFunction(tools)
+          try {
-            .then(finalChunk => {
+            const finalChunk = await optionsArg.finalFunction(tools);
             if (finalChunk) {
               controller.enqueue(finalChunk);
             }
-            })  
+          } catch (err) {
-            .catch(err => controller.error(err))
+            controller.error(err);
-            .finally(() => controller.terminate());
+          } finally {
+            controller.terminate();
+          }
         } else {
           controller.terminate();
         }
-      }
+      },
     });
 
     this.options = optionsArg;
+
+    // Start producing data if readFunction is provided
+    if (this.options.readFunction) {
+      this._startReading();
+    }
   }
 
-  // Method to create a custom readable stream that integrates the readFunction
+  private async _startReading() {
-  // readFunction is executed whenever the stream is being read from and nothing is enqueued
+    const writable = this.writable;
-  getCustomReadableStream() {
+    const writer = writable.getWriter();
-    const readableStream = this.readable;
-    const options = this.options;
-    const customReadable = new ReadableStream({
-      async pull(controller) {
-        const reader = readableStream.getReader();
 
-        // Check the current state of the original stream
+    const tools: IStreamToolsRead<TInput, TOutput> = {
-        const { value, done } = await reader.read();
+      done: () => writer.close(),
-        reader.releaseLock();
+      write: async (writeArg) => await writer.write(writeArg),
+    };
 
-        if (done) {
+    try {
-          // If the original stream is done, close the custom readable stream
+      await this.options.readFunction(tools);
-          controller.close();
+    } catch (err) {
-        } else {
+      writer.abort(err);
-          if (value) {
+    } finally {
-            // If there is data in the original stream, enqueue it and do not execute the readFunction
+      writer.releaseLock();
-            controller.enqueue(value);
+    }
-          } else if (options.readFunction) {
+  }
-            // If the original stream is empty, execute the readFunction and read again
+
-            await options.readFunction({
+  // Static method example (adjust as needed)
-              done: () => controller.close(),
+  static fromUInt8Array(uint8Array: Uint8Array): WebDuplexStream<Uint8Array, Uint8Array> {
-              write: (writeArg) => controller.enqueue(writeArg),
+    const stream = new WebDuplexStream<Uint8Array, Uint8Array>({
+      writeFunction: async (chunk, { push }) => {
+        push(chunk); // Directly push the chunk as is
+        return null;
+      },
     });
 
-            const newReader = readableStream.getReader();
+    const writer = stream.writable.getWriter();
-            const { value: newValue, done: newDone } = await newReader.read();
+    writer.write(uint8Array).then(() => writer.close());
-            newReader.releaseLock();
 
-            if (newDone) {
+    return stream;
-              controller.close();
-            } else {
-              controller.enqueue(newValue);
-            }
-          }
-        }
-      }
-    });
-
-    return customReadable;
   }
 }