Compare commits

...

4 Commits

6 changed files with 43 additions and 6 deletions

View File

@ -1,5 +1,17 @@
# Changelog
## 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.1",
"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.1',
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,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>;
@ -80,11 +105,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.1',
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.'
}