mirror of
https://github.com/filecoin-project/slate.git
synced 2024-12-26 18:44:56 +03:00
56 lines
1.4 KiB
JavaScript
56 lines
1.4 KiB
JavaScript
const { Transform } = require("stream");
|
|
|
|
/**
|
|
* Transform stream with throttle functionality
|
|
*
|
|
* @class
|
|
* @extends Transform
|
|
* @param {Object} options - Configuration options
|
|
* @param {Integer} options.bytes - Number of bytes to send in one chunk
|
|
* @param {Integer} options.interval - Interval for sending chunks, in miliseconds
|
|
*/
|
|
export default class Throttle extends Transform {
|
|
constructor(options) {
|
|
super();
|
|
Object.assign(this, options);
|
|
this.previousPassTime = Date.now();
|
|
this.queue = [];
|
|
this.intervalId = this.setPushInterval();
|
|
}
|
|
|
|
_transform(chunks, _, cb) {
|
|
for (const chunk of chunks) {
|
|
this.queue.push(chunk);
|
|
}
|
|
this.isQueueFull() ? setTimeout(cb, this.interval) : cb();
|
|
}
|
|
|
|
_flush(cb) {
|
|
clearInterval(this.intervalId);
|
|
this.intervalId = this.setPushInterval(cb);
|
|
}
|
|
|
|
setPushInterval(cb = null) {
|
|
return setInterval(() => {
|
|
const elapsedTime = Date.now() - this.previousPassTime;
|
|
if (elapsedTime < this.interval) return;
|
|
|
|
if (this.queue.length > 0) {
|
|
this.push(this.getChunk());
|
|
return (this.previousPassTime += elapsedTime);
|
|
}
|
|
|
|
clearInterval(this.intervalId);
|
|
return cb && cb();
|
|
}, this.interval / 10);
|
|
}
|
|
|
|
getChunk() {
|
|
return Buffer.from(this.queue.splice(0, this.bytes));
|
|
}
|
|
|
|
isQueueFull() {
|
|
return this.queue.length >= 2 * this.bytes;
|
|
}
|
|
}
|