Compare commits

...

6 Commits

6 changed files with 43 additions and 6 deletions

View File

@ -1,5 +1,23 @@
# Changelog
## 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

View File

@ -1,6 +1,6 @@
{
"name": "@push.rocks/smartstream",
"version": "3.1.2",
"version": "3.2.2",
"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",

View File

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

View File

@ -52,6 +52,25 @@ 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();
const { value, done } = await reader.read();
if (value !== undefined) {
smartDuplex.push(value);
}
reader.releaseLock();
if (done) {
smartDuplex.end();
}
},
});
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 +99,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();

View File

@ -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> {

View File

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