113 lines
3.9 KiB
TypeScript
113 lines
3.9 KiB
TypeScript
import * as plugins from './smartstream.plugins.js';
|
|
import { Duplex, type DuplexOptions } from 'stream';
|
|
export interface SmartStreamOptions<TInput, TOutput> extends DuplexOptions {
|
|
// You can add more custom options relevant to TInput and TOutput if necessary
|
|
}
|
|
|
|
export class SmartStream<TInput = any, TOutput = any> extends Duplex {
|
|
private observableSubscription?: plugins.smartrx.rxjs.Subscription;
|
|
private asyncChunkModifier?: (chunk: TInput) => Promise<TOutput>;
|
|
|
|
constructor(options?: SmartStreamOptions<TInput, TOutput>, asyncChunkModifierArg?: (chunk: TInput) => Promise<TOutput>) {
|
|
super(options);
|
|
this.asyncChunkModifier = asyncChunkModifierArg;
|
|
}
|
|
|
|
// Ensure the _write method types the chunk as TInput and encodes TOutput
|
|
public async _write(chunk: TInput, encoding: string, callback: (error?: Error | null) => void) {
|
|
try {
|
|
if (this.asyncChunkModifier) {
|
|
const modifiedChunk = await this.asyncChunkModifier(chunk);
|
|
if (!this.push(modifiedChunk)) {
|
|
// Handle backpressure here if necessary
|
|
}
|
|
} else {
|
|
if (!this.push(chunk as unknown as TOutput)) {
|
|
// Handle backpressure here if necessary
|
|
}
|
|
}
|
|
callback();
|
|
} catch (err) {
|
|
callback(err);
|
|
}
|
|
}
|
|
|
|
static fromBuffer(buffer: Buffer, options?: DuplexOptions): SmartStream {
|
|
const smartStream = new SmartStream(options);
|
|
process.nextTick(() => {
|
|
smartStream.push(buffer);
|
|
smartStream.push(null); // Signal the end of the data
|
|
});
|
|
return smartStream;
|
|
}
|
|
|
|
static fromObservable(observable: plugins.smartrx.rxjs.Observable<any>, options?: DuplexOptions): SmartStream {
|
|
const smartStream = new SmartStream(options);
|
|
smartStream.observableSubscription = observable.subscribe({
|
|
next: (data) => {
|
|
if (!smartStream.push(data)) {
|
|
// Pause the observable if the stream buffer is full
|
|
smartStream.observableSubscription?.unsubscribe();
|
|
smartStream.once('drain', () => {
|
|
// Resume the observable when the stream buffer is drained
|
|
smartStream.observableSubscription?.unsubscribe();
|
|
smartStream.observableSubscription = observable.subscribe(data => {
|
|
smartStream.push(data);
|
|
});
|
|
});
|
|
}
|
|
},
|
|
error: (err) => {
|
|
smartStream.emit('error', err);
|
|
},
|
|
complete: () => {
|
|
smartStream.push(null); // Signal the end of the data
|
|
}
|
|
});
|
|
|
|
return smartStream;
|
|
}
|
|
|
|
static fromReplaySubject(replaySubject: plugins.smartrx.rxjs.ReplaySubject<any>, options?: DuplexOptions): SmartStream {
|
|
const smartStream = new SmartStream(options);
|
|
let isBackpressured = false;
|
|
|
|
// Subscribe to the ReplaySubject
|
|
const subscription = replaySubject.subscribe({
|
|
next: (data) => {
|
|
const canPush = smartStream.push(data);
|
|
if (!canPush) {
|
|
// If push returns false, pause the subscription because of backpressure
|
|
isBackpressured = true;
|
|
subscription.unsubscribe();
|
|
}
|
|
},
|
|
error: (err) => {
|
|
smartStream.emit('error', err);
|
|
},
|
|
complete: () => {
|
|
smartStream.push(null); // End the stream when the ReplaySubject completes
|
|
}
|
|
});
|
|
|
|
// Listen for 'drain' event to resume the subscription if it was paused
|
|
smartStream.on('drain', () => {
|
|
if (isBackpressured) {
|
|
isBackpressured = false;
|
|
// Resubscribe to the ReplaySubject since we previously paused
|
|
smartStream.observableSubscription = replaySubject.subscribe({
|
|
next: (data) => {
|
|
if (!smartStream.push(data)) {
|
|
smartStream.observableSubscription?.unsubscribe();
|
|
isBackpressured = true;
|
|
}
|
|
},
|
|
// No need to repeat error and complete handling here because it's already set up above
|
|
});
|
|
}
|
|
});
|
|
|
|
return smartStream;
|
|
}
|
|
}
|