fix(core): update
This commit is contained in:
@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@push.rocks/smartstream',
|
||||
version: '2.0.4',
|
||||
version: '2.0.5',
|
||||
description: 'simplifies access to node streams'
|
||||
}
|
||||
|
@ -1,3 +1,4 @@
|
||||
export * from './smartstream.classes.smartstream.js';
|
||||
export * from './smartstream.classes.streamwrapper.js';
|
||||
export * from './smartstream.classes.streamintake.js';
|
||||
export * from './smartstream.duplex.js';
|
55
ts/smartstream.classes.smartstream.ts
Normal file
55
ts/smartstream.classes.smartstream.ts
Normal file
@ -0,0 +1,55 @@
|
||||
import * as plugins from './smartstream.plugins.js';
|
||||
import { Duplex, type DuplexOptions } from 'stream';
|
||||
|
||||
export class SmartStream extends Duplex {
|
||||
private observableSubscription?: plugins.smartrx.rxjs.Subscription;
|
||||
|
||||
constructor(options?: DuplexOptions) {
|
||||
super(options);
|
||||
}
|
||||
|
||||
_read(size: number) {
|
||||
// Implement if you need custom behavior, otherwise leave it empty
|
||||
}
|
||||
|
||||
_write(chunk: any, encoding: string, callback: (error?: Error | null) => void) {
|
||||
// Implement if you need custom behavior
|
||||
callback();
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
Reference in New Issue
Block a user