41 lines
626 B
JavaScript
41 lines
626 B
JavaScript
|
import {Transform} from 'node:stream';
|
||
|
|
|
||
|
|
class ObjectTransform extends Transform {
|
||
|
|
constructor() {
|
||
|
|
super({
|
||
|
|
objectMode: true,
|
||
|
|
});
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
export class FilterStream extends ObjectTransform {
|
||
|
|
constructor(filter) {
|
||
|
|
super();
|
||
|
|
this._filter = filter;
|
||
|
|
}
|
||
|
|
|
||
|
|
_transform(data, encoding, callback) {
|
||
|
|
if (this._filter(data)) {
|
||
|
|
this.push(data);
|
||
|
|
}
|
||
|
|
|
||
|
|
callback();
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
export class UniqueStream extends ObjectTransform {
|
||
|
|
constructor() {
|
||
|
|
super();
|
||
|
|
this._pushed = new Set();
|
||
|
|
}
|
||
|
|
|
||
|
|
_transform(data, encoding, callback) {
|
||
|
|
if (!this._pushed.has(data)) {
|
||
|
|
this.push(data);
|
||
|
|
this._pushed.add(data);
|
||
|
|
}
|
||
|
|
|
||
|
|
callback();
|
||
|
|
}
|
||
|
|
}
|