fix(WebDuplexStream): Improved read/write interface and error handling in WebDuplexStream
This commit is contained in:
parent
04c22f73df
commit
7b1d2199e9
@ -1,5 +1,13 @@
|
||||
# Changelog
|
||||
|
||||
## 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
|
||||
|
||||
|
||||
@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@push.rocks/smartstream',
|
||||
version: '3.1.0',
|
||||
version: '3.1.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.'
|
||||
}
|
||||
|
||||
@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@push.rocks/smartstream',
|
||||
version: '3.1.0',
|
||||
version: '3.1.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.'
|
||||
}
|
||||
|
||||
@ -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 WebDuplexStream is being read from
|
||||
* and at the same time if nothing is enqueued
|
||||
* The read function is called when data needs to be read into the stream.
|
||||
*/
|
||||
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.
|
||||
* It is called anytime a chunk is written to the stream.
|
||||
* The write function is called whenever 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,12 +41,90 @@ export interface WebDuplexStreamOptions<TInput, TOutput> {
|
||||
}
|
||||
|
||||
export class WebDuplexStream<TInput = any, TOutput = any> extends TransformStream<TInput, TOutput> {
|
||||
// INSTANCE
|
||||
options: WebDuplexStreamOptions<TInput, TOutput>;
|
||||
|
||||
constructor(optionsArg: WebDuplexStreamOptions<TInput, TOutput>) {
|
||||
super({
|
||||
async start(controller) {
|
||||
// Optionally initialize any state here
|
||||
},
|
||||
async transform(chunk, controller) {
|
||||
if (optionsArg?.writeFunction) {
|
||||
const tools: IStreamToolsWrite<TInput, TOutput> = {
|
||||
truncate: () => controller.terminate(),
|
||||
push: (pushArg: TOutput) => controller.enqueue(pushArg),
|
||||
};
|
||||
|
||||
try {
|
||||
const writeReturnChunk = await optionsArg.writeFunction(chunk, tools);
|
||||
if (writeReturnChunk !== undefined && writeReturnChunk !== null) {
|
||||
controller.enqueue(writeReturnChunk);
|
||||
}
|
||||
} catch (err) {
|
||||
controller.error(err);
|
||||
}
|
||||
} else {
|
||||
// If no writeFunction is provided, pass the chunk through
|
||||
controller.enqueue(chunk as unknown as TOutput);
|
||||
}
|
||||
},
|
||||
async flush(controller) {
|
||||
if (optionsArg?.finalFunction) {
|
||||
const tools: IStreamToolsWrite<TInput, TOutput> = {
|
||||
truncate: () => controller.terminate(),
|
||||
push: (pushArg) => controller.enqueue(pushArg),
|
||||
};
|
||||
|
||||
try {
|
||||
const finalChunk = await optionsArg.finalFunction(tools);
|
||||
if (finalChunk) {
|
||||
controller.enqueue(finalChunk);
|
||||
}
|
||||
} catch (err) {
|
||||
controller.error(err);
|
||||
} finally {
|
||||
controller.terminate();
|
||||
}
|
||||
} else {
|
||||
controller.terminate();
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
this.options = optionsArg;
|
||||
|
||||
// Start producing data if readFunction is provided
|
||||
if (this.options.readFunction) {
|
||||
this._startReading();
|
||||
}
|
||||
}
|
||||
|
||||
private async _startReading() {
|
||||
const writable = this.writable;
|
||||
const writer = writable.getWriter();
|
||||
|
||||
const tools: IStreamToolsRead<TInput, TOutput> = {
|
||||
done: () => writer.close(),
|
||||
write: async (writeArg) => await writer.write(writeArg),
|
||||
};
|
||||
|
||||
try {
|
||||
await this.options.readFunction(tools);
|
||||
} catch (err) {
|
||||
writer.abort(err);
|
||||
} finally {
|
||||
writer.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
// Static method example (adjust as needed)
|
||||
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();
|
||||
@ -58,99 +132,4 @@ export class WebDuplexStream<TInput = any, TOutput = any> extends TransformStrea
|
||||
|
||||
return stream;
|
||||
}
|
||||
|
||||
// INSTANCE
|
||||
options: WebDuplexStreamOptions<TInput, TOutput>;
|
||||
|
||||
constructor(optionsArg: WebDuplexStreamOptions<TInput, TOutput>) {
|
||||
// here we call into the official web stream api
|
||||
super({
|
||||
async transform(chunk, controller) {
|
||||
// Transformation logic remains unchanged
|
||||
if (optionsArg?.writeFunction) {
|
||||
const tools: IStreamToolsWrite<TInput, TOutput> = {
|
||||
truncate: () => controller.terminate(),
|
||||
push: (pushArg: TOutput) => controller.enqueue(pushArg),
|
||||
};
|
||||
|
||||
try {
|
||||
const writeReturnChunk = await optionsArg.writeFunction(chunk, tools);
|
||||
if (writeReturnChunk) { // return chunk is optional
|
||||
controller.enqueue(writeReturnChunk);
|
||||
}
|
||||
} catch (err) {
|
||||
controller.error(err);
|
||||
}
|
||||
} else {
|
||||
controller.error(new Error('No write function provided'));
|
||||
}
|
||||
},
|
||||
async flush(controller) {
|
||||
// Flush logic remains unchanged
|
||||
if (optionsArg?.finalFunction) {
|
||||
const tools: IStreamToolsWrite<TInput, TOutput> = {
|
||||
truncate: () => controller.terminate(),
|
||||
push: (pipeObject) => controller.enqueue(pipeObject),
|
||||
};
|
||||
|
||||
optionsArg.finalFunction(tools)
|
||||
.then(finalChunk => {
|
||||
if (finalChunk) {
|
||||
controller.enqueue(finalChunk);
|
||||
}
|
||||
})
|
||||
.catch(err => controller.error(err))
|
||||
.finally(() => controller.terminate());
|
||||
} else {
|
||||
controller.terminate();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
this.options = optionsArg;
|
||||
}
|
||||
|
||||
// Method to create a custom readable stream that integrates the readFunction
|
||||
// readFunction is executed whenever the stream is being read from and nothing is enqueued
|
||||
getCustomReadableStream() {
|
||||
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 { value, done } = await reader.read();
|
||||
reader.releaseLock();
|
||||
|
||||
if (done) {
|
||||
// If the original stream is done, close the custom readable stream
|
||||
controller.close();
|
||||
} else {
|
||||
if (value) {
|
||||
// If there is data in the original stream, enqueue it and do not execute the readFunction
|
||||
controller.enqueue(value);
|
||||
} else if (options.readFunction) {
|
||||
// If the original stream is empty, execute the readFunction and read again
|
||||
await options.readFunction({
|
||||
done: () => controller.close(),
|
||||
write: (writeArg) => controller.enqueue(writeArg),
|
||||
});
|
||||
|
||||
const newReader = readableStream.getReader();
|
||||
const { value: newValue, done: newDone } = await newReader.read();
|
||||
newReader.releaseLock();
|
||||
|
||||
if (newDone) {
|
||||
controller.close();
|
||||
} else {
|
||||
controller.enqueue(newValue);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
return customReadable;
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user