3.2.5

changelog.md

@@ -1,5 +1,45 @@
 # Changelog
 
+## 2024-11-19 - 3.2.5 - fix(nodewebhelpers)
+Fix import and use correct module structure for Node.js streams in smartstream.nodewebhelpers.ts
+
+- Corrected the import statement for the fs module.
+- Updated the use of the fs.createReadStream method.
+
+## 2024-10-16 - 3.2.4 - fix(SmartDuplex)
+Fix stream termination when reading from a web readable stream
+
+- Resolved an issue in SmartDuplex where the stream did not properly terminate after reaching the end of a web readable stream.
+
+## 2024-10-16 - 3.2.3 - fix(smartduplex)
+Enhance documentation for read function in SmartDuplex
+
+- Added inline comments to clarify the behavior and importance of unlocking the reader in the readFunction of SmartDuplex.fromWebReadableStream.
+
+## 2024-10-16 - 3.2.2 - fix(SmartDuplex)
+Fix issue with SmartDuplex fromWebReadableStream method
+
+- Resolved a potential unhandled promise rejection in fromWebReadableStream method
+- Ensured proper release of stream reader lock in case of read completion
+
+## 2024-10-16 - 3.2.1 - fix(core)
+Fix the order of operations in SmartDuplex _read method to ensure proper waiting for items.
+
+- Adjusted the order of reading function execution and waiting for items in the SmartDuplex _read method.
+- Fixed potential issues with stream data processing timing.
+
+## 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
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartstream",
-  "version": "3.1.1",
+  "version": "3.2.5",
   "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.1.1',
+  version: '3.2.5',
   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/smartstream.classes.smartduplex.ts

@@ -52,6 +52,29 @@ 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>({
+      /**
+       * this function is called whenever the stream is being read from and at the same time if nothing is enqueued
+       * therefor it is important to always unlock the reader after reading
+       */
+      readFunction: async () => {
+        const reader = readableStream.getReader();
+        const { value, done } = await reader.read();
+        if (value !== undefined) {
+          smartDuplex.push(value);
+        }
+        reader.releaseLock();
+        if (done) {
+          smartDuplex.push(null);
+        }
+      },
+    });
+    return smartDuplex;
+  }
+
   // INSTANCE
   private backpressuredArray: plugins.lik.BackpressuredArray<TOutput>; // an array that only takes a defined amount of items
   public options: ISmartDuplexOptions<TInput, TOutput>;
@@ -80,11 +103,11 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
 
   public async _read(size: number): Promise<void> {
     this.debugLog(`${this.options.name}: read was called`);
-    await this.backpressuredArray.waitForItems();
-    this.debugLog(`${this.options.name}: successfully waited for items.`);
     if (this.options.readFunction) {
       await this.options.readFunction();
     }
+    await this.backpressuredArray.waitForItems();
+    this.debugLog(`${this.options.name}: successfully waited for items.`);
     let canPushMore = true;
     while (this.backpressuredArray.data.length > 0 && canPushMore) {
       const nextChunk = this.backpressuredArray.shift();

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

@@ -1,4 +1,4 @@
-import { createReadStream } from 'fs';
+import * as plugins from './smartstream.plugins.js';
 
 /**
  * Creates a Web ReadableStream from a file.
@@ -7,7 +7,7 @@ import { createReadStream } from 'fs';
  * @returns A Web ReadableStream that reads the file in chunks
  */
 export function createWebReadableStreamFromFile(filePath: string): ReadableStream<Uint8Array> {
-  const fileStream = createReadStream(filePath);
+  const fileStream = plugins.fs.createReadStream(filePath);
 
   return new ReadableStream({
     start(controller) {
@@ -31,4 +31,121 @@ export function createWebReadableStreamFromFile(filePath: string): ReadableStrea
       fileStream.destroy();
     }
   });
+}
+
+/**
+ * Converts a Web ReadableStream to a Node.js Readable stream.
+ *
+ * @param webStream - The Web ReadableStream to convert
+ * @returns A Node.js Readable stream that reads data from the Web ReadableStream
+ */
+export function convertWebReadableToNodeReadable(webStream: ReadableStream<Uint8Array>): plugins.stream.Readable {
+  const reader = webStream.getReader();
+
+  return new plugins.stream.Readable({
+    async read() {
+      try {
+        const { value, done } = await reader.read();
+        if (done) {
+          this.push(null); // Signal end of stream
+        } else {
+          this.push(Buffer.from(value)); // Convert Uint8Array to Buffer for Node.js Readable
+        }
+      } catch (err) {
+        this.destroy(err); // Handle errors by destroying the stream
+      }
+    }
+  });
+}
+
+/**
+ * Converts a Node.js Readable stream to a Web ReadableStream.
+ *
+ * @param nodeStream - The Node.js Readable stream to convert
+ * @returns A Web ReadableStream that reads data from the Node.js Readable stream
+ */
+export function convertNodeReadableToWebReadable(nodeStream: plugins.stream.Readable): ReadableStream<Uint8Array> {
+  return new ReadableStream({
+    start(controller) {
+      nodeStream.on('data', (chunk) => {
+        controller.enqueue(new Uint8Array(chunk));
+      });
+
+      nodeStream.on('end', () => {
+        controller.close();
+      });
+
+      nodeStream.on('error', (err) => {
+        controller.error(err);
+      });
+    },
+    cancel() {
+      nodeStream.destroy();
+    }
+  });
+}
+
+/**
+ * Converts a Web WritableStream to a Node.js Writable stream.
+ *
+ * @param webWritable - The Web WritableStream to convert
+ * @returns A Node.js Writable stream that writes data to the Web WritableStream
+ */
+export function convertWebWritableToNodeWritable(webWritable: WritableStream<Uint8Array>): plugins.stream.Writable {
+  const writer = webWritable.getWriter();
+
+  return new plugins.stream.Writable({
+    async write(chunk, encoding, callback) {
+      try {
+        await writer.write(new Uint8Array(chunk));
+        callback();
+      } catch (err) {
+        callback(err);
+      }
+    },
+    final(callback) {
+      writer.close().then(() => callback()).catch(callback);
+    },
+    destroy(err, callback) {
+      writer.abort(err).then(() => callback(err)).catch(callback);
+    }
+  });
+}
+
+/**
+ * Converts a Node.js Writable stream to a Web WritableStream.
+ *
+ * @param nodeWritable - The Node.js Writable stream to convert
+ * @returns A Web WritableStream that writes data to the Node.js Writable stream
+ */
+export function convertNodeWritableToWebWritable(nodeWritable: plugins.stream.Writable): WritableStream<Uint8Array> {
+  return new WritableStream({
+    write(chunk) {
+      return new Promise((resolve, reject) => {
+        nodeWritable.write(Buffer.from(chunk), (err) => {
+          if (err) {
+            reject(err);
+          } else {
+            resolve();
+          }
+        });
+      });
+    },
+    close() {
+      return new Promise((resolve, reject) => {
+        nodeWritable.end((err) => {
+          if (err) {
+            reject(err);
+          } else {
+            resolve();
+          }
+        });
+      });
+    },
+    abort(reason) {
+      return new Promise((resolve, reject) => {
+        nodeWritable.destroy(reason);
+      });
+    }
+  });
 }

ts/smartstream.plugins.ts

@@ -1,7 +1,8 @@
 // node native
+import * as fs from 'fs';
 import * as stream from 'stream';
 
-export { stream };
+export { fs, stream };
 
 // pushrocks scope
 import * as lik from '@push.rocks/lik';

ts_web/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartstream',
-  version: '3.1.1',
+  version: '3.2.5',
   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.'
 }