feat(smartduplex): improve backpressure handling and web/node stream interoperability
This commit is contained in:
@@ -56,67 +56,116 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
|
||||
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);
|
||||
}
|
||||
},
|
||||
objectMode: true,
|
||||
});
|
||||
|
||||
// Acquire reader ONCE
|
||||
const reader = readableStream.getReader();
|
||||
let reading = false;
|
||||
|
||||
// Override _read to pull from the web reader
|
||||
smartDuplex._read = function (_size: number) {
|
||||
if (reading) return;
|
||||
reading = true;
|
||||
reader.read().then(
|
||||
({ value, done }) => {
|
||||
reading = false;
|
||||
if (done) {
|
||||
smartDuplex.push(null);
|
||||
} else {
|
||||
smartDuplex.push(value);
|
||||
}
|
||||
},
|
||||
(err) => {
|
||||
reading = false;
|
||||
smartDuplex.destroy(err);
|
||||
}
|
||||
);
|
||||
};
|
||||
|
||||
// Cancel reader on destroy
|
||||
smartDuplex.on('close', () => {
|
||||
reader.cancel().catch(() => {});
|
||||
});
|
||||
|
||||
return smartDuplex;
|
||||
}
|
||||
|
||||
// INSTANCE
|
||||
private backpressuredArray: plugins.lik.BackpressuredArray<TOutput>; // an array that only takes a defined amount of items
|
||||
private backpressuredArray: plugins.lik.BackpressuredArray<TOutput>;
|
||||
public options: ISmartDuplexOptions<TInput, TOutput>;
|
||||
private observableSubscription?: plugins.smartrx.rxjs.Subscription;
|
||||
private _consumerWantsData = false;
|
||||
private _readFunctionRunning = false;
|
||||
|
||||
private debugLog(messageArg: string) {
|
||||
// optional debug log
|
||||
if (this.options.debug) {
|
||||
console.log(messageArg);
|
||||
}
|
||||
}
|
||||
|
||||
constructor(optionsArg?: ISmartDuplexOptions<TInput, TOutput>) {
|
||||
const safeOptions = optionsArg || {} as ISmartDuplexOptions<TInput, TOutput>;
|
||||
super(
|
||||
Object.assign(
|
||||
{
|
||||
highWaterMark: 1,
|
||||
},
|
||||
optionsArg
|
||||
safeOptions
|
||||
)
|
||||
);
|
||||
this.options = optionsArg;
|
||||
this.options = safeOptions;
|
||||
this.backpressuredArray = new plugins.lik.BackpressuredArray<TOutput>(
|
||||
this.options.highWaterMark || 1
|
||||
);
|
||||
}
|
||||
|
||||
public async _read(size: number): Promise<void> {
|
||||
this.debugLog(`${this.options.name}: read was called`);
|
||||
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) {
|
||||
/**
|
||||
* Synchronously drains items from the backpressuredArray into the readable side.
|
||||
* Stops when push() returns false (consumer is full) or array is empty.
|
||||
*/
|
||||
private _drainBackpressuredArray(): void {
|
||||
while (this.backpressuredArray.data.length > 0) {
|
||||
const nextChunk = this.backpressuredArray.shift();
|
||||
canPushMore = this.push(nextChunk);
|
||||
if (nextChunk === null) {
|
||||
// EOF signal — push null to end readable side
|
||||
this.push(null);
|
||||
this._consumerWantsData = false;
|
||||
return;
|
||||
}
|
||||
const canPushMore = this.push(nextChunk);
|
||||
if (!canPushMore) {
|
||||
this._consumerWantsData = false;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// _read must NOT be async — Node.js ignores the return value
|
||||
public _read(size: number): void {
|
||||
this.debugLog(`${this.options.name}: read was called`);
|
||||
this._consumerWantsData = true;
|
||||
|
||||
// Drain any buffered items first
|
||||
if (this.backpressuredArray.data.length > 0) {
|
||||
this._drainBackpressuredArray();
|
||||
}
|
||||
|
||||
// If readFunction exists and is not already running, start it
|
||||
if (this.options.readFunction && !this._readFunctionRunning) {
|
||||
this._readFunctionRunning = true;
|
||||
this.options.readFunction().then(
|
||||
() => { this._readFunctionRunning = false; },
|
||||
(err) => { this._readFunctionRunning = false; this.destroy(err); }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
public async backpressuredPush(pushArg: TOutput) {
|
||||
const canPushMore = this.backpressuredArray.push(pushArg);
|
||||
// Try to drain if the consumer wants data
|
||||
if (this._consumerWantsData) {
|
||||
this._drainBackpressuredArray();
|
||||
}
|
||||
if (!canPushMore) {
|
||||
this.debugLog(`${this.options.name}: cannot push more`);
|
||||
await this.backpressuredArray.waitForSpace();
|
||||
@@ -126,83 +175,151 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
|
||||
}
|
||||
|
||||
private asyncWritePromiseObjectmap = new plugins.lik.ObjectMap<Promise<any>>();
|
||||
// Ensure the _write method types the chunk as TInput and encodes TOutput
|
||||
public async _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) {
|
||||
|
||||
// _write must NOT be async — Node.js ignores the return value
|
||||
public _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) {
|
||||
if (!this.options.writeFunction) {
|
||||
return callback(new Error('No stream function provided'));
|
||||
}
|
||||
|
||||
let callbackCalled = false;
|
||||
const safeCallback = (err?: Error | null) => {
|
||||
if (!callbackCalled) {
|
||||
callbackCalled = true;
|
||||
callback(err);
|
||||
}
|
||||
};
|
||||
|
||||
let isTruncated = false;
|
||||
const tools: IStreamTools = {
|
||||
truncate: () => {
|
||||
this.push(null);
|
||||
isTruncated = true;
|
||||
callback();
|
||||
safeCallback();
|
||||
this.push(null);
|
||||
},
|
||||
push: async (pushArg: TOutput) => {
|
||||
return await this.backpressuredPush(pushArg);
|
||||
},
|
||||
};
|
||||
|
||||
try {
|
||||
const writeDeferred = plugins.smartpromise.defer();
|
||||
this.asyncWritePromiseObjectmap.add(writeDeferred.promise);
|
||||
const modifiedChunk = await this.options.writeFunction(chunk, tools);
|
||||
if (isTruncated) {
|
||||
return;
|
||||
}
|
||||
if (modifiedChunk) {
|
||||
await tools.push(modifiedChunk);
|
||||
}
|
||||
callback();
|
||||
writeDeferred.resolve();
|
||||
writeDeferred.promise.then(() => {
|
||||
const writeDeferred = plugins.smartpromise.defer();
|
||||
this.asyncWritePromiseObjectmap.add(writeDeferred.promise);
|
||||
|
||||
this.options.writeFunction(chunk, tools).then(
|
||||
(modifiedChunk) => {
|
||||
if (isTruncated) {
|
||||
writeDeferred.resolve();
|
||||
this.asyncWritePromiseObjectmap.remove(writeDeferred.promise);
|
||||
return;
|
||||
}
|
||||
const finish = () => {
|
||||
safeCallback();
|
||||
writeDeferred.resolve();
|
||||
this.asyncWritePromiseObjectmap.remove(writeDeferred.promise);
|
||||
};
|
||||
if (modifiedChunk !== undefined && modifiedChunk !== null) {
|
||||
this.backpressuredPush(modifiedChunk).then(finish, (err) => {
|
||||
safeCallback(err);
|
||||
writeDeferred.resolve();
|
||||
this.asyncWritePromiseObjectmap.remove(writeDeferred.promise);
|
||||
});
|
||||
} else {
|
||||
finish();
|
||||
}
|
||||
},
|
||||
(err) => {
|
||||
safeCallback(err);
|
||||
writeDeferred.resolve();
|
||||
this.asyncWritePromiseObjectmap.remove(writeDeferred.promise);
|
||||
});
|
||||
} catch (err) {
|
||||
callback(err);
|
||||
}
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
public async _final(callback: (error?: Error | null) => void) {
|
||||
await Promise.all(this.asyncWritePromiseObjectmap.getArray());
|
||||
if (this.options.finalFunction) {
|
||||
const tools: IStreamTools = {
|
||||
truncate: () => callback(),
|
||||
push: async (pipeObject) => {
|
||||
return this.backpressuredArray.push(pipeObject);
|
||||
},
|
||||
};
|
||||
|
||||
try {
|
||||
const finalChunk = await this.options.finalFunction(tools);
|
||||
if (finalChunk) {
|
||||
this.backpressuredArray.push(finalChunk);
|
||||
}
|
||||
} catch (err) {
|
||||
this.backpressuredArray.push(null);
|
||||
// _final must NOT be async — Node.js ignores the return value
|
||||
public _final(callback: (error?: Error | null) => void) {
|
||||
let callbackCalled = false;
|
||||
const safeCallback = (err?: Error | null) => {
|
||||
if (!callbackCalled) {
|
||||
callbackCalled = true;
|
||||
callback(err);
|
||||
return;
|
||||
}
|
||||
}
|
||||
this.backpressuredArray.push(null);
|
||||
callback();
|
||||
};
|
||||
|
||||
Promise.all(this.asyncWritePromiseObjectmap.getArray()).then(() => {
|
||||
if (this.options.finalFunction) {
|
||||
const tools: IStreamTools = {
|
||||
truncate: () => safeCallback(),
|
||||
push: async (pipeObject) => {
|
||||
return await this.backpressuredPush(pipeObject);
|
||||
},
|
||||
};
|
||||
|
||||
this.options.finalFunction(tools).then(
|
||||
(finalChunk) => {
|
||||
const pushNull = () => {
|
||||
this.backpressuredArray.push(null);
|
||||
if (this._consumerWantsData) {
|
||||
this._drainBackpressuredArray();
|
||||
}
|
||||
safeCallback();
|
||||
};
|
||||
|
||||
if (finalChunk !== undefined && finalChunk !== null) {
|
||||
this.backpressuredPush(finalChunk).then(pushNull, (err) => {
|
||||
safeCallback(err);
|
||||
});
|
||||
} else {
|
||||
pushNull();
|
||||
}
|
||||
},
|
||||
(err) => {
|
||||
this.backpressuredArray.push(null);
|
||||
if (this._consumerWantsData) {
|
||||
this._drainBackpressuredArray();
|
||||
}
|
||||
safeCallback(err);
|
||||
}
|
||||
);
|
||||
} else {
|
||||
this.backpressuredArray.push(null);
|
||||
if (this._consumerWantsData) {
|
||||
this._drainBackpressuredArray();
|
||||
}
|
||||
safeCallback();
|
||||
}
|
||||
}, (err) => {
|
||||
safeCallback(err);
|
||||
});
|
||||
}
|
||||
|
||||
public async getWebStreams(): Promise<{ readable: ReadableStream; writable: WritableStream }> {
|
||||
const duplex = this;
|
||||
let readableClosed = false;
|
||||
|
||||
const readable = new ReadableStream({
|
||||
start(controller) {
|
||||
duplex.on('readable', () => {
|
||||
const onReadable = () => {
|
||||
let chunk;
|
||||
while (null !== (chunk = duplex.read())) {
|
||||
controller.enqueue(chunk);
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
duplex.on('end', () => {
|
||||
controller.close();
|
||||
});
|
||||
const onEnd = () => {
|
||||
if (!readableClosed) {
|
||||
readableClosed = true;
|
||||
controller.close();
|
||||
}
|
||||
cleanup();
|
||||
};
|
||||
|
||||
const cleanup = () => {
|
||||
duplex.removeListener('readable', onReadable);
|
||||
duplex.removeListener('end', onEnd);
|
||||
};
|
||||
|
||||
duplex.on('readable', onReadable);
|
||||
duplex.on('end', onEnd);
|
||||
},
|
||||
cancel(reason) {
|
||||
duplex.destroy(new Error(reason));
|
||||
@@ -212,22 +329,38 @@ export class SmartDuplex<TInput = any, TOutput = any> extends Duplex {
|
||||
const writable = new WritableStream({
|
||||
write(chunk) {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
let resolved = false;
|
||||
const onDrain = () => {
|
||||
if (!resolved) {
|
||||
resolved = true;
|
||||
resolve();
|
||||
}
|
||||
};
|
||||
|
||||
const isBackpressured = !duplex.write(chunk, (error) => {
|
||||
if (error) {
|
||||
reject(error);
|
||||
} else {
|
||||
if (!resolved) {
|
||||
resolved = true;
|
||||
duplex.removeListener('drain', onDrain);
|
||||
reject(error);
|
||||
}
|
||||
} else if (!isBackpressured && !resolved) {
|
||||
resolved = true;
|
||||
resolve();
|
||||
}
|
||||
});
|
||||
|
||||
if (isBackpressured) {
|
||||
duplex.once('drain', resolve);
|
||||
duplex.once('drain', onDrain);
|
||||
}
|
||||
});
|
||||
},
|
||||
close() {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
duplex.end(resolve);
|
||||
duplex.end((err: Error | null) => {
|
||||
if (err) reject(err);
|
||||
else resolve();
|
||||
});
|
||||
});
|
||||
},
|
||||
abort(reason) {
|
||||
|
||||
Reference in New Issue
Block a user