opusdev/vector-similarity-api

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