| 1 | 'use strict';
|
|---|
| 2 |
|
|---|
| 3 | import stream from 'stream';
|
|---|
| 4 | import utils from '../utils.js';
|
|---|
| 5 |
|
|---|
| 6 | const kInternals = Symbol('internals');
|
|---|
| 7 |
|
|---|
| 8 | class AxiosTransformStream extends stream.Transform {
|
|---|
| 9 | constructor(options) {
|
|---|
| 10 | options = utils.toFlatObject(
|
|---|
| 11 | options,
|
|---|
| 12 | {
|
|---|
| 13 | maxRate: 0,
|
|---|
| 14 | chunkSize: 64 * 1024,
|
|---|
| 15 | minChunkSize: 100,
|
|---|
| 16 | timeWindow: 500,
|
|---|
| 17 | ticksRate: 2,
|
|---|
| 18 | samplesCount: 15,
|
|---|
| 19 | },
|
|---|
| 20 | null,
|
|---|
| 21 | (prop, source) => {
|
|---|
| 22 | return !utils.isUndefined(source[prop]);
|
|---|
| 23 | }
|
|---|
| 24 | );
|
|---|
| 25 |
|
|---|
| 26 | super({
|
|---|
| 27 | readableHighWaterMark: options.chunkSize,
|
|---|
| 28 | });
|
|---|
| 29 |
|
|---|
| 30 | const internals = (this[kInternals] = {
|
|---|
| 31 | timeWindow: options.timeWindow,
|
|---|
| 32 | chunkSize: options.chunkSize,
|
|---|
| 33 | maxRate: options.maxRate,
|
|---|
| 34 | minChunkSize: options.minChunkSize,
|
|---|
| 35 | bytesSeen: 0,
|
|---|
| 36 | isCaptured: false,
|
|---|
| 37 | notifiedBytesLoaded: 0,
|
|---|
| 38 | ts: Date.now(),
|
|---|
| 39 | bytes: 0,
|
|---|
| 40 | onReadCallback: null,
|
|---|
| 41 | });
|
|---|
| 42 |
|
|---|
| 43 | this.on('newListener', (event) => {
|
|---|
| 44 | if (event === 'progress') {
|
|---|
| 45 | if (!internals.isCaptured) {
|
|---|
| 46 | internals.isCaptured = true;
|
|---|
| 47 | }
|
|---|
| 48 | }
|
|---|
| 49 | });
|
|---|
| 50 | }
|
|---|
| 51 |
|
|---|
| 52 | _read(size) {
|
|---|
| 53 | const internals = this[kInternals];
|
|---|
| 54 |
|
|---|
| 55 | if (internals.onReadCallback) {
|
|---|
| 56 | internals.onReadCallback();
|
|---|
| 57 | }
|
|---|
| 58 |
|
|---|
| 59 | return super._read(size);
|
|---|
| 60 | }
|
|---|
| 61 |
|
|---|
| 62 | _transform(chunk, encoding, callback) {
|
|---|
| 63 | const internals = this[kInternals];
|
|---|
| 64 | const maxRate = internals.maxRate;
|
|---|
| 65 |
|
|---|
| 66 | const readableHighWaterMark = this.readableHighWaterMark;
|
|---|
| 67 |
|
|---|
| 68 | const timeWindow = internals.timeWindow;
|
|---|
| 69 |
|
|---|
| 70 | const divider = 1000 / timeWindow;
|
|---|
| 71 | const bytesThreshold = maxRate / divider;
|
|---|
| 72 | const minChunkSize =
|
|---|
| 73 | internals.minChunkSize !== false
|
|---|
| 74 | ? Math.max(internals.minChunkSize, bytesThreshold * 0.01)
|
|---|
| 75 | : 0;
|
|---|
| 76 |
|
|---|
| 77 | const pushChunk = (_chunk, _callback) => {
|
|---|
| 78 | const bytes = Buffer.byteLength(_chunk);
|
|---|
| 79 | internals.bytesSeen += bytes;
|
|---|
| 80 | internals.bytes += bytes;
|
|---|
| 81 |
|
|---|
| 82 | internals.isCaptured && this.emit('progress', internals.bytesSeen);
|
|---|
| 83 |
|
|---|
| 84 | if (this.push(_chunk)) {
|
|---|
| 85 | process.nextTick(_callback);
|
|---|
| 86 | } else {
|
|---|
| 87 | internals.onReadCallback = () => {
|
|---|
| 88 | internals.onReadCallback = null;
|
|---|
| 89 | process.nextTick(_callback);
|
|---|
| 90 | };
|
|---|
| 91 | }
|
|---|
| 92 | };
|
|---|
| 93 |
|
|---|
| 94 | const transformChunk = (_chunk, _callback) => {
|
|---|
| 95 | const chunkSize = Buffer.byteLength(_chunk);
|
|---|
| 96 | let chunkRemainder = null;
|
|---|
| 97 | let maxChunkSize = readableHighWaterMark;
|
|---|
| 98 | let bytesLeft;
|
|---|
| 99 | let passed = 0;
|
|---|
| 100 |
|
|---|
| 101 | if (maxRate) {
|
|---|
| 102 | const now = Date.now();
|
|---|
| 103 |
|
|---|
| 104 | if (!internals.ts || (passed = now - internals.ts) >= timeWindow) {
|
|---|
| 105 | internals.ts = now;
|
|---|
| 106 | bytesLeft = bytesThreshold - internals.bytes;
|
|---|
| 107 | internals.bytes = bytesLeft < 0 ? -bytesLeft : 0;
|
|---|
| 108 | passed = 0;
|
|---|
| 109 | }
|
|---|
| 110 |
|
|---|
| 111 | bytesLeft = bytesThreshold - internals.bytes;
|
|---|
| 112 | }
|
|---|
| 113 |
|
|---|
| 114 | if (maxRate) {
|
|---|
| 115 | if (bytesLeft <= 0) {
|
|---|
| 116 | // next time window
|
|---|
| 117 | return setTimeout(() => {
|
|---|
| 118 | _callback(null, _chunk);
|
|---|
| 119 | }, timeWindow - passed);
|
|---|
| 120 | }
|
|---|
| 121 |
|
|---|
| 122 | if (bytesLeft < maxChunkSize) {
|
|---|
| 123 | maxChunkSize = bytesLeft;
|
|---|
| 124 | }
|
|---|
| 125 | }
|
|---|
| 126 |
|
|---|
| 127 | if (maxChunkSize && chunkSize > maxChunkSize && chunkSize - maxChunkSize > minChunkSize) {
|
|---|
| 128 | chunkRemainder = _chunk.subarray(maxChunkSize);
|
|---|
| 129 | _chunk = _chunk.subarray(0, maxChunkSize);
|
|---|
| 130 | }
|
|---|
| 131 |
|
|---|
| 132 | pushChunk(
|
|---|
| 133 | _chunk,
|
|---|
| 134 | chunkRemainder
|
|---|
| 135 | ? () => {
|
|---|
| 136 | process.nextTick(_callback, null, chunkRemainder);
|
|---|
| 137 | }
|
|---|
| 138 | : _callback
|
|---|
| 139 | );
|
|---|
| 140 | };
|
|---|
| 141 |
|
|---|
| 142 | transformChunk(chunk, function transformNextChunk(err, _chunk) {
|
|---|
| 143 | if (err) {
|
|---|
| 144 | return callback(err);
|
|---|
| 145 | }
|
|---|
| 146 |
|
|---|
| 147 | if (_chunk) {
|
|---|
| 148 | transformChunk(_chunk, transformNextChunk);
|
|---|
| 149 | } else {
|
|---|
| 150 | callback(null);
|
|---|
| 151 | }
|
|---|
| 152 | });
|
|---|
| 153 | }
|
|---|
| 154 | }
|
|---|
| 155 |
|
|---|
| 156 | export default AxiosTransformStream;
|
|---|