Compare commits
6 Commits
Author | SHA1 | Date | |
---|---|---|---|
f7e9636bf6 | |||
f211cc8ddd | |||
60c8824f33 | |||
40e8e06ff1 | |||
30f2facd59 | |||
ddb7d4af03 |
18
changelog.md
18
changelog.md
@ -1,5 +1,23 @@
|
|||||||
# Changelog
|
# 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)
|
## 2024-10-14 - 3.1.2 - fix(WebDuplexStream)
|
||||||
Fix variable naming inconsistency in WebDuplexStream test
|
Fix variable naming inconsistency in WebDuplexStream test
|
||||||
|
|
||||||
|
@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "@push.rocks/smartstream",
|
"name": "@push.rocks/smartstream",
|
||||||
"version": "3.1.2",
|
"version": "3.2.2",
|
||||||
"private": false,
|
"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.",
|
"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",
|
"type": "module",
|
||||||
|
@ -3,6 +3,6 @@
|
|||||||
*/
|
*/
|
||||||
export const commitinfo = {
|
export const commitinfo = {
|
||||||
name: '@push.rocks/smartstream',
|
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.'
|
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.'
|
||||||
}
|
}
|
||||||
|
@ -52,6 +52,25 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
|
|||||||
return smartDuplex;
|
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
|
// INSTANCE
|
||||||
private backpressuredArray: plugins.lik.BackpressuredArray<TOutput>; // an array that only takes a defined amount of items
|
private backpressuredArray: plugins.lik.BackpressuredArray<TOutput>; // an array that only takes a defined amount of items
|
||||||
public options: ISmartDuplexOptions<TInput, TOutput>;
|
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> {
|
public async _read(size: number): Promise<void> {
|
||||||
this.debugLog(`${this.options.name}: read was called`);
|
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) {
|
if (this.options.readFunction) {
|
||||||
await this.options.readFunction();
|
await this.options.readFunction();
|
||||||
}
|
}
|
||||||
|
await this.backpressuredArray.waitForItems();
|
||||||
|
this.debugLog(`${this.options.name}: successfully waited for items.`);
|
||||||
let canPushMore = true;
|
let canPushMore = true;
|
||||||
while (this.backpressuredArray.data.length > 0 && canPushMore) {
|
while (this.backpressuredArray.data.length > 0 && canPushMore) {
|
||||||
const nextChunk = this.backpressuredArray.shift();
|
const nextChunk = this.backpressuredArray.shift();
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
import { Transform, type TransformCallback, type TransformOptions } from 'stream';
|
import { type TransformOptions } from 'stream';
|
||||||
import { SmartDuplex } from './smartstream.classes.smartduplex.js';
|
import { SmartDuplex } from './smartstream.classes.smartduplex.js';
|
||||||
|
|
||||||
export interface AsyncTransformFunction<TInput, TOutput> {
|
export interface AsyncTransformFunction<TInput, TOutput> {
|
||||||
|
@ -3,6 +3,6 @@
|
|||||||
*/
|
*/
|
||||||
export const commitinfo = {
|
export const commitinfo = {
|
||||||
name: '@push.rocks/smartstream',
|
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.'
|
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.'
|
||||||
}
|
}
|
||||||
|
Reference in New Issue
Block a user