Source: lib/Transport.js

  1. /*
  2.  * Copyright OpenSearch Contributors
  3.  * SPDX-License-Identifier: Apache-2.0
  4.  *
  5.  * The OpenSearch Contributors require contributions made to
  6.  * this file be licensed under the Apache-2.0 license or a
  7.  * compatible open source license.
  8.  *
  9.  */
  11. /*
  12.  * Licensed to Elasticsearch B.V. under one or more contributor
  13.  * license agreements. See the NOTICE file distributed with
  14.  * this work for additional information regarding copyright
  15.  * ownership. Elasticsearch B.V. licenses this file to you under
  16.  * the Apache License, Version 2.0 (the "License"); you may
  17.  * not use this file except in compliance with the License.
  18.  * You may obtain a copy of the License at
  19.  *
  20.  *    http://www.apache.org/licenses/LICENSE-2.0
  21.  *
  22.  * Unless required by applicable law or agreed to in writing,
  23.  * software distributed under the License is distributed on an
  24.  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  25.  * KIND, either express or implied.  See the License for the
  26.  * specific language governing permissions and limitations
  27.  * under the License.
  28.  */
  30. 'use strict';
  32. const debug = require('debug')('opensearch');
  33. const os = require('os');
  34. const { gzip, unzip, createGzip } = require('zlib');
  35. const buffer = require('buffer');
  36. const ms = require('ms');
  37. const { EventEmitter } = require('events');
  38. const {
  39.   ConnectionError,
  40.   RequestAbortedError,
  41.   NoLivingConnectionsError,
  42.   ResponseError,
  43.   ConfigurationError,
  44. } = require('./errors');
  46. const noop = () => {};
  48. const compatibleCheckEmitter = new EventEmitter();
  49. const clientVersion = require('../package.json').version;
  50. const userAgent = `opensearch-js/${clientVersion} (${os.platform()} ${os.release()}-${os.arch()}; Node.js ${
  51.   process.version
  52. })`;
  53. const MAX_BUFFER_LENGTH = buffer.constants.MAX_LENGTH;
  54. const MAX_STRING_LENGTH = buffer.constants.MAX_STRING_LENGTH;
  55. const HEAP_SIZE_LIMIT = require('v8').getHeapStatistics().heap_size_limit;
  56. const kCompatibleCheck = Symbol('compatible check');
  57. const kApiVersioning = Symbol('api versioning');
  59. /** Default Transport Layer */
  60. class Transport {
  61.   constructor(opts) {
  62.     if (typeof opts.compression === 'string' && opts.compression !== 'gzip') {
  63.       throw new ConfigurationError(`Invalid compression: '${opts.compression}'`);
  64.     }
  66.     this.emit = opts.emit;
  67.     this.connectionPool = opts.connectionPool;
  68.     this.serializer = opts.serializer;
  69.     this.maxRetries = opts.maxRetries;
  70.     this.requestTimeout = toMs(opts.requestTimeout);
  71.     this.suggestCompression = opts.suggestCompression === true;
  72.     this.compression = opts.compression || false;
  73.     this.context = opts.context || null;
  74.     this.headers = Object.assign(
  75.       {},
  76.       { 'user-agent': userAgent },
  77.       opts.suggestCompression === true ? { 'accept-encoding': 'gzip,deflate' } : null,
  78.       lowerCaseHeaders(opts.headers)
  79.     );
  80.     this.sniffInterval = opts.sniffInterval;
  81.     this.sniffOnConnectionFault = opts.sniffOnConnectionFault;
  82.     this.sniffEndpoint = opts.sniffEndpoint;
  83.     this.generateRequestId = opts.generateRequestId || generateRequestId();
  84.     this.name = opts.name;
  85.     this.opaqueIdPrefix = opts.opaqueIdPrefix;
  86.     this[kCompatibleCheck] = 0; // 0 = to be checked, 1 = checking, 2 = checked-ok, 3 checked-notok
  87.     this[kApiVersioning] = process.env.OPENSEARCH_CLIENT_APIVERSIONING === 'true';
  88.     this.memoryCircuitBreaker = opts.memoryCircuitBreaker;
  90.     this.nodeFilter = opts.nodeFilter || defaultNodeFilter;
  91.     if (typeof opts.nodeSelector === 'function') {
  92.       this.nodeSelector = opts.nodeSelector;
  93.     } else if (opts.nodeSelector === 'round-robin') {
  94.       this.nodeSelector = roundRobinSelector();
  95.     } else if (opts.nodeSelector === 'random') {
  96.       this.nodeSelector = randomSelector;
  97.     } else {
  98.       this.nodeSelector = roundRobinSelector();
  99.     }
  101.     this._sniffEnabled = typeof this.sniffInterval === 'number';
  102.     this._nextSniff = this._sniffEnabled ? Date.now() + this.sniffInterval : 0;
  103.     this._isSniffing = false;
  104.     this._auth = opts.auth;
  106.     if (opts.sniffOnStart === true) {
  107.       // timer needed otherwise it will clash
  108.       // with the compatible check testing
  109.       setTimeout(() => {
  110.         this.sniff({ reason: Transport.sniffReasons.SNIFF_ON_START });
  111.       }, 10);
  112.     }
  113.   }
  115.   /**
  116.    * @param {Object} params
  117.    * @param {string} params.method - HTTP Method (e.g. HEAD, GET, POST...)
  118.    * @param {string} params.path - Relative URL path
  119.    * @param {Object | string} [params.body] - Body of a standard request.
  120.    * @param {Object[] | string} [params.bulkBody] - Body of a bulk request.
  121.    * @param {Object[] | string} [params.querystring] - Querystring params.
  122.    *
  123.    * @param {Object} options
  124.    * @param {number} [options.id] - Request ID
  125.    * @param {Object} [options.context] - Object used for observability
  126.    * @param {number} [options.maxRetries] - Max number of retries
  127.    * @param {false | 'gzip'} [options.compression] - Request body compression, if any
  128.    * @param {boolean} [options.asStream] - Whether to emit the response as stream
  129.    * @param {number[]} [options.ignore] - Response's Error Status Codes to ignore
  130.    * @param {Object} [options.headers] - Request headers
  131.    * @param {Object | string} [options.querystring] - Request's query string
  132.    * @param {number} [options.requestTimeout] - Max request timeout in milliseconds
  133.    *
  134.    * @param {function} callback - Callback that handles errors and response
  135.    */
  136.   request(params, options, callback) {
  137.     options = options || {};
  138.     if (typeof options === 'function') {
  139.       callback = options;
  140.       options = {};
  141.     }
  142.     let p = null;
  144.     // promises support
  145.     if (callback === undefined) {
  146.       let onFulfilled = null;
  147.       let onRejected = null;
  148.       p = new Promise((resolve, reject) => {
  149.         onFulfilled = resolve;
  150.         onRejected = reject;
  151.       });
  152.       callback = function callback(err, result) {
  153.         err ? onRejected(err) : onFulfilled(result);
  154.       };
  155.     }
  157.     const meta = {
  158.       context: null,
  159.       request: {
  160.         params: null,
  161.         options: null,
  162.         id: options.id || this.generateRequestId(params, options),
  163.       },
  164.       name: this.name,
  165.       connection: null,
  166.       attempts: 0,
  167.       aborted: false,
  168.     };
  170.     if (this.context != null && options.context != null) {
  171.       meta.context = Object.assign({}, this.context, options.context);
  172.     } else if (this.context != null) {
  173.       meta.context = this.context;
  174.     } else if (options.context != null) {
  175.       meta.context = options.context;
  176.     }
  178.     const result = {
  179.       body: null,
  180.       statusCode: null,
  181.       headers: null,
  182.       meta,
  183.     };
  185.     Object.defineProperty(result, 'warnings', {
  186.       get() {
  187.         return this.headers && this.headers.warning
  188.           ? this.headers.warning.split(/(?!\B"[^"]*),(?![^"]*"\B)/)
  189.           : null;
  190.       },
  191.     });
  193.     // We should not retry if we are sending a stream body, because we should store in memory
  194.     // a copy of the stream to be able to send it again, but since we don't know in advance
  195.     // the size of the stream, we risk to take too much memory.
  196.     // Furthermore, copying everytime the stream is very a expensive operation.
  197.     const maxRetries =
  198.       isStream(params.body) || isStream(params.bulkBody)
  199.         ? 0
  200.         : typeof options.maxRetries === 'number'
  201.           ? options.maxRetries
  202.           : this.maxRetries;
  203.     const compression = options.compression !== undefined ? options.compression : this.compression;
  204.     let request = { abort: noop };
  205.     const transportReturn = {
  206.       then(onFulfilled, onRejected) {
  207.         if (p != null) {
  208.           return p.then(onFulfilled, onRejected);
  209.         }
  210.       },
  211.       catch(onRejected) {
  212.         if (p != null) {
  213.           return p.catch(onRejected);
  214.         }
  215.       },
  216.       abort() {
  217.         meta.aborted = true;
  218.         request.abort();
  219.         debug('Aborting request', params);
  220.         return this;
  221.       },
  222.       finally(onFinally) {
  223.         if (p != null) {
  224.           return p.finally(onFinally);
  225.         }
  226.       },
  227.     };
  229.     const makeRequest = () => {
  230.       if (meta.aborted === true) {
  231.         return process.nextTick(callback, new RequestAbortedError(), result);
  232.       }
  233.       meta.connection = this.getConnection({ requestId: meta.request.id });
  234.       if (meta.connection == null) {
  235.         return process.nextTick(callback, new NoLivingConnectionsError(), result);
  236.       }
  237.       this.emit('request', null, result);
  238.       // perform the actual http request
  239.       request = meta.connection.request(params, onResponse);
  240.     };
  242.     const onConnectionError = (err) => {
  243.       if (err.name !== 'RequestAbortedError') {
  244.         // if there is an error in the connection
  245.         // let's mark the connection as dead
  246.         this.connectionPool.markDead(meta.connection);
  248.         if (this.sniffOnConnectionFault === true) {
  249.           this.sniff({
  250.             reason: Transport.sniffReasons.SNIFF_ON_CONNECTION_FAULT,
  251.             requestId: meta.request.id,
  252.           });
  253.         }
  255.         // retry logic
  256.         if (meta.attempts < maxRetries) {
  257.           meta.attempts++;
  258.           debug(`Retrying request, there are still ${maxRetries - meta.attempts} attempts`, params);
  259.           makeRequest();
  260.           return;
  261.         }
  262.       }
  264.       err.meta = result;
  265.       this.emit('response', err, result);
  266.       return callback(err, result);
  267.     };
  269.     const onResponse = (err, response) => {
  270.       if (err !== null) {
  271.         return onConnectionError(err);
  272.       }
  274.       result.statusCode = response.statusCode;
  275.       result.headers = response.headers;
  277.       if (options.asStream === true) {
  278.         result.body = response;
  279.         this.emit('response', null, result);
  280.         callback(null, result);
  281.         return;
  282.       }
  284.       const contentEncoding = (result.headers['content-encoding'] || '').toLowerCase();
  285.       const isCompressed =
  286.         contentEncoding.indexOf('gzip') > -1 || contentEncoding.indexOf('deflate') > -1;
  287.       /* istanbul ignore else */
  288.       if (result.headers['content-length'] !== undefined) {
  289.         const contentLength = Number(result.headers['content-length']);
  290.         // nodeJS data type limit check
  291.         if (isCompressed && contentLength > MAX_BUFFER_LENGTH) {
  292.           response.destroy();
  293.           return onConnectionError(
  294.             new RequestAbortedError(
  295.               `The content length (${contentLength}) is bigger than the maximum allowed buffer (${MAX_BUFFER_LENGTH})`,
  296.               result
  297.             )
  298.           );
  299.         } else if (contentLength > MAX_STRING_LENGTH) {
  300.           response.destroy();
  301.           return onConnectionError(
  302.             new RequestAbortedError(
  303.               `The content length (${contentLength}) is bigger than the maximum allowed string (${MAX_STRING_LENGTH})`,
  304.               result
  305.             )
  306.           );
  307.         } else if (shouldApplyCircuitBreaker(contentLength)) {
  308.           // Abort this response to avoid OOM crash of dashboards.
  309.           response.destroy();
  310.           return onConnectionError(
  311.             new RequestAbortedError(
  312.               `The content length (${contentLength}) is bigger than the maximum allowed heap memory limit.`,
  313.               result
  314.             )
  315.           );
  316.         }
  317.       }
  318.       // if the response is compressed, we must handle it
  319.       // as buffer for allowing decompression later
  320.       let payload = isCompressed ? [] : '';
  321.       const onData = isCompressed
  322.         ? (chunk) => {
  323.             payload.push(chunk);
  324.           }
  325.         : (chunk) => {
  326.             payload += chunk;
  327.           };
  328.       const onEnd = (err) => {
  329.         response.removeListener('data', onData);
  330.         response.removeListener('end', onEnd);
  331.         response.removeListener('error', onEnd);
  332.         response.removeListener('aborted', onAbort);
  334.         if (err) {
  335.           return onConnectionError(new ConnectionError(err.message));
  336.         }
  338.         if (isCompressed) {
  339.           unzip(Buffer.concat(payload), onBody);
  340.         } else {
  341.           onBody(null, payload);
  342.         }
  343.       };
  345.       const onAbort = () => {
  346.         response.destroy();
  347.         onEnd(new Error('Response aborted while reading the body'));
  348.       };
  350.       if (!isCompressed) {
  351.         response.setEncoding('utf8');
  352.       }
  354.       this.emit('deserialization', null, result);
  355.       response.on('data', onData);
  356.       response.on('error', onEnd);
  357.       response.on('end', onEnd);
  358.       response.on('aborted', onAbort);
  359.     };
  360.     // Helper function to check if memory circuit breaker enabled and the response payload is too large to fit into available heap memory.
  361.     const shouldApplyCircuitBreaker = (contentLength) => {
  362.       if (!this.memoryCircuitBreaker || !this.memoryCircuitBreaker.enabled) return false;
  363.       const maxPercentage = validateMemoryPercentage(this.memoryCircuitBreaker.maxPercentage);
  364.       const heapUsed = process.memoryUsage().heapUsed;
  365.       return contentLength + heapUsed > HEAP_SIZE_LIMIT * maxPercentage;
  366.     };
  368.     const onBody = (err, payload) => {
  369.       if (err) {
  370.         this.emit('response', err, result);
  371.         return callback(err, result);
  372.       }
  373.       if (Buffer.isBuffer(payload)) {
  374.         payload = payload.toString();
  375.       }
  376.       const isHead = params.method === 'HEAD';
  377.       // we should attempt the payload deserialization only if:
  378.       //    - a `content-type` is defined and is equal to `application/json`
  379.       //    - the request is not a HEAD request
  380.       //    - the payload is not an empty string
  381.       if (
  382.         result.headers['content-type'] !== undefined &&
  383.         (result.headers['content-type'].indexOf('application/json') > -1 ||
  384.           result.headers['content-type'].indexOf('application/vnd.opensearch+json') > -1) &&
  385.         isHead === false &&
  386.         payload !== ''
  387.       ) {
  388.         try {
  389.           result.body = this.serializer.deserialize(payload);
  390.         } catch (err) {
  391.           this.emit('response', err, result);
  392.           return callback(err, result);
  393.         }
  394.       } else {
  395.         // cast to boolean if the request method was HEAD and there was no error
  396.         result.body = isHead === true && result.statusCode < 400 ? true : payload;
  397.       }
  399.       // we should ignore the statusCode if the user has configured the `ignore` field with
  400.       // the statusCode we just got or if the request method is HEAD and the statusCode is 404
  401.       const ignoreStatusCode =
  402.         (Array.isArray(options.ignore) && options.ignore.indexOf(result.statusCode) > -1) ||
  403.         (isHead === true && result.statusCode === 404);
  405.       if (
  406.         ignoreStatusCode === false &&
  407.         (result.statusCode === 502 || result.statusCode === 503 || result.statusCode === 504)
  408.       ) {
  409.         // if the statusCode is 502/3/4 we should run our retry strategy
  410.         // and mark the connection as dead
  411.         this.connectionPool.markDead(meta.connection);
  412.         // retry logic (we should not retry on "429 - Too Many Requests")
  413.         if (meta.attempts < maxRetries && result.statusCode !== 429) {
  414.           meta.attempts++;
  415.           debug(`Retrying request, there are still ${maxRetries - meta.attempts} attempts`, params);
  416.           makeRequest();
  417.           return;
  418.         }
  419.       } else {
  420.         // everything has worked as expected, let's mark
  421.         // the connection as alive (or confirm it)
  422.         this.connectionPool.markAlive(meta.connection);
  423.       }
  425.       if (ignoreStatusCode === false && result.statusCode >= 400) {
  426.         const error = new ResponseError(result);
  427.         this.emit('response', error, result);
  428.         callback(error, result);
  429.       } else {
  430.         // cast to boolean if the request method was HEAD
  431.         if (isHead === true && result.statusCode === 404) {
  432.           result.body = false;
  433.         }
  434.         this.emit('response', null, result);
  435.         callback(null, result);
  436.       }
  437.     };
  439.     const prepareRequest = () => {
  440.       this.emit('serialization', null, result);
  441.       const headers = Object.assign({}, this.headers, lowerCaseHeaders(options.headers));
  443.       if (options.opaqueId !== undefined) {
  444.         headers['x-opaque-id'] =
  445.           this.opaqueIdPrefix !== null ? this.opaqueIdPrefix + options.opaqueId : options.opaqueId;
  446.       }
  448.       // handle json body
  449.       if (params.body != null) {
  450.         if (shouldSerialize(params.body) === true) {
  451.           try {
  452.             params.body = this.serializer.serialize(params.body);
  453.           } catch (err) {
  454.             this.emit('request', err, result);
  455.             process.nextTick(callback, err, result);
  456.             return transportReturn;
  457.           }
  458.         }
  460.         if (params.body !== '') {
  461.           headers['content-type'] =
  462.             headers['content-type'] ||
  463.             (this[kApiVersioning]
  464.               ? 'application/vnd.opensearch+json; compatible-with=7'
  465.               : 'application/json');
  466.         }
  468.         // handle ndjson body
  469.       } else if (params.bulkBody != null) {
  470.         if (shouldSerialize(params.bulkBody) === true) {
  471.           try {
  472.             params.body = this.serializer.ndserialize(params.bulkBody);
  473.           } catch (err) {
  474.             this.emit('request', err, result);
  475.             process.nextTick(callback, err, result);
  476.             return transportReturn;
  477.           }
  478.         } else {
  479.           params.body = params.bulkBody;
  480.         }
  481.         if (params.body !== '') {
  482.           headers['content-type'] =
  483.             headers['content-type'] ||
  484.             (this[kApiVersioning]
  485.               ? 'application/vnd.opensearch+x-ndjson; compatible-with=7'
  486.               : 'application/x-ndjson');
  487.         }
  488.       }
  490.       params.headers = headers;
  491.       // serializes the querystring
  492.       if (options.querystring == null) {
  493.         params.querystring = this.serializer.qserialize(params.querystring);
  494.       } else {
  495.         params.querystring = this.serializer.qserialize(
  496.           Object.assign({}, params.querystring, options.querystring)
  497.         );
  498.       }
  499.       if (this._auth !== null && typeof this._auth === 'object' && 'credentials' in this._auth) {
  500.         params.auth = this._auth;
  501.       }
  503.       // handles request timeout
  504.       params.timeout = toMs(options.requestTimeout || this.requestTimeout);
  505.       if (options.asStream === true) params.asStream = true;
  506.       meta.request.params = params;
  507.       meta.request.options = options;
  509.       // handle compression
  510.       if (params.body !== '' && params.body != null) {
  511.         if (isStream(params.body) === true) {
  512.           if (compression === 'gzip') {
  513.             params.headers['content-encoding'] = compression;
  514.             params.body = params.body.pipe(createGzip());
  515.           }
  516.           makeRequest();
  517.         } else if (compression === 'gzip') {
  518.           gzip(params.body, (err, buffer) => {
  519.             /* istanbul ignore next */
  520.             if (err) {
  521.               this.emit('request', err, result);
  522.               return callback(err, result);
  523.             }
  524.             params.headers['content-encoding'] = compression;
  525.             params.headers['content-length'] = '' + Buffer.byteLength(buffer);
  526.             params.body = buffer;
  527.             makeRequest();
  528.           });
  529.         } else {
  530.           params.headers['content-length'] = '' + Buffer.byteLength(params.body);
  531.           makeRequest();
  532.         }
  533.       } else {
  534.         makeRequest();
  535.       }
  536.     };
  538.     prepareRequest();
  540.     return transportReturn;
  541.   }
  543.   getConnection(opts) {
  544.     const now = Date.now();
  545.     if (this._sniffEnabled === true && now > this._nextSniff) {
  546.       this.sniff({ reason: Transport.sniffReasons.SNIFF_INTERVAL, requestId: opts.requestId });
  547.     }
  548.     return this.connectionPool.getConnection({
  549.       filter: this.nodeFilter,
  550.       selector: this.nodeSelector,
  551.       requestId: opts.requestId,
  552.       name: this.name,
  553.       now,
  554.     });
  555.   }
  557.   sniff(opts, callback = noop) {
  558.     if (this._isSniffing === true) return;
  559.     this._isSniffing = true;
  560.     debug('Started sniffing request');
  562.     if (typeof opts === 'function') {
  563.       callback = opts;
  564.       opts = { reason: Transport.sniffReasons.DEFAULT };
  565.     }
  567.     const { reason } = opts;
  569.     const request = {
  570.       method: 'GET',
  571.       path: this.sniffEndpoint,
  572.     };
  574.     this.request(request, { id: opts.requestId }, (err, result) => {
  575.       this._isSniffing = false;
  576.       if (this._sniffEnabled === true) {
  577.         this._nextSniff = Date.now() + this.sniffInterval;
  578.       }
  580.       if (err != null) {
  581.         debug('Sniffing errored', err);
  582.         result.meta.sniff = { hosts: [], reason };
  583.         this.emit('sniff', err, result);
  584.         return callback(err);
  585.       }
  587.       debug('Sniffing ended successfully', result.body);
  588.       const protocol = result.meta.connection.url.protocol || /* istanbul ignore next */ 'http:';
  589.       const hosts = this.connectionPool.nodesToHost(result.body.nodes, protocol);
  590.       this.connectionPool.update(hosts);
  592.       result.meta.sniff = { hosts, reason };
  593.       this.emit('sniff', null, result);
  594.       callback(null, hosts);
  595.     });
  596.   }
  598.   // checkCompatibleInfo validates whether the informations are compatible
  599.   checkCompatibleInfo() {
  600.     debug('Start compatible check');
  601.     this[kCompatibleCheck] = 1;
  602.     this.request(
  603.       {
  604.         method: 'GET',
  605.         path: '/',
  606.       },
  607.       (err, result) => {
  608.         this[kCompatibleCheck] = 3;
  609.         if (err) {
  610.           debug('compatible check failed', err);
  611.           if (err.statusCode === 401 || err.statusCode === 403) {
  612.             this[kCompatibleCheck] = 2;
  613.             process.emitWarning(
  614.               'The client is unable to verify the distribution due to security privileges on the server side. Some functionality may not be compatible if the server is running an unsupported product.'
  615.             );
  616.             compatibleCheckEmitter.emit('compatible-check', true);
  617.           } else {
  618.             this[kCompatibleCheck] = 0;
  619.             compatibleCheckEmitter.emit('compatible-check', false);
  620.           }
  621.         } else {
  622.           debug('Checking OpenSearch version', result.body, result.headers);
  623.           if (result.body.version == null || typeof result.body.version.number !== 'string') {
  624.             debug("Can't access OpenSearch version");
  625.             return compatibleCheckEmitter.emit('compatible-check', false);
  626.           }
  628.           const distribution = result.body.version.distribution;
  629.           const version = result.body.version.number.split('.');
  630.           const major = Number(version[0]);
  632.           // support OpenSearch validation
  633.           if (distribution === 'opensearch') {
  634.             debug('Valid OpenSearch distribution');
  635.             this[kCompatibleCheck] = 2;
  636.             return compatibleCheckEmitter.emit('compatible-check', true);
  637.           }
  639.           // support odfe > v7 validation
  640.           if (major !== 7) {
  641.             debug('Invalid distribution');
  642.             return compatibleCheckEmitter.emit('compatible-check', false);
  643.           }
  645.           debug('Valid OpenSearch distribution');
  646.           this[kCompatibleCheck] = 2;
  647.           compatibleCheckEmitter.emit('compatible-check', true);
  648.         }
  649.       }
  650.     );
  651.   }
  652. }
  654. Transport.sniffReasons = {
  655.   SNIFF_ON_START: 'sniff-on-start',
  656.   SNIFF_INTERVAL: 'sniff-interval',
  657.   SNIFF_ON_CONNECTION_FAULT: 'sniff-on-connection-fault',
  658.   // TODO: find a better name
  659.   DEFAULT: 'default',
  660. };
  662. function toMs(time) {
  663.   if (typeof time === 'string') {
  664.     return ms(time);
  665.   }
  666.   return time;
  667. }
  669. function shouldSerialize(obj) {
  670.   return (
  671.     typeof obj !== 'string' && typeof obj.pipe !== 'function' && Buffer.isBuffer(obj) === false
  672.   );
  673. }
  675. function isStream(obj) {
  676.   return obj != null && typeof obj.pipe === 'function';
  677. }
  679. function defaultNodeFilter(node) {
  680.   // avoid cluster_manager or master only nodes
  681.   // TODO: remove role check on master when master is not supported
  682.   if (
  683.     (node.roles.cluster_manager === true || node.roles.master === true) &&
  684.     node.roles.data === false &&
  685.     node.roles.ingest === false
  686.   ) {
  687.     return false;
  688.   }
  689.   return true;
  690. }
  692. function roundRobinSelector() {
  693.   let current = -1;
  694.   return function _roundRobinSelector(connections) {
  695.     if (++current >= connections.length) {
  696.       current = 0;
  697.     }
  698.     return connections[current];
  699.   };
  700. }
  702. function randomSelector(connections) {
  703.   const index = Math.floor(Math.random() * connections.length);
  704.   return connections[index];
  705. }
  707. function generateRequestId() {
  708.   const maxInt = 2147483647;
  709.   let nextReqId = 0;
  710.   return function genReqId() {
  711.     return (nextReqId = (nextReqId + 1) & maxInt);
  712.   };
  713. }
  715. function lowerCaseHeaders(oldHeaders) {
  716.   if (oldHeaders == null) return oldHeaders;
  717.   const newHeaders = {};
  718.   for (const header in oldHeaders) {
  719.     newHeaders[header.toLowerCase()] = oldHeaders[header];
  720.   }
  721.   return newHeaders;
  722. }
  724. function validateMemoryPercentage(percentage) {
  725.   if (percentage < 0 || percentage > 1) return 1.0;
  726.   return percentage;
  727. }
  729. module.exports = Transport;
  730. module.exports.internals = {
  731.   defaultNodeFilter,
  732.   roundRobinSelector,
  733.   randomSelector,
  734.   generateRequestId,
  735.   lowerCaseHeaders,
  736.   toMs,
  737. };