From c9635c4a71ebe79ac9bba2b00d8066d6dc3d4166 Mon Sep 17 00:00:00 2001 From: delvedor Date: Tue, 30 Oct 2018 16:32:10 +0100 Subject: [PATCH] WIP: initial prototype - Added support for different format of requestTimemout - Changed api method result - Now we are always returning the result in case of error - Improved body deserialization - Added cast to boolen for HEAD requests - Added support for already serialized strings in the ndserializer - Fixed qserializer in case of null object - Updated Errors --- index.js | 4 ++-- lib/Serializer.js | 7 +++++- lib/Transport.js | 61 ++++++++++++++++++++++++++++++----------------- lib/errors.js | 30 ++++++++++++----------- package.json | 1 + 5 files changed, 64 insertions(+), 39 deletions(-) diff --git a/index.js b/index.js index 02dc24202..25cf37cff 100644 --- a/index.js +++ b/index.js @@ -7,7 +7,7 @@ const ConnectionPool = require('./lib/ConnectionPool') const Serializer = require('./lib/Serializer') const selectors = require('./lib/Selectors') const symbols = require('./lib/symbols') -const { BadConfigurationError } = require('./lib/errors') +const { ConfigurationError } = require('./lib/errors') // const buildApi = require('../monorepo/packages/es-api-6') @@ -22,7 +22,7 @@ class Client extends EventEmitter { constructor (opts = {}) { super() if (!opts.host) { - throw new BadConfigurationError('Missing host option') + throw new ConfigurationError('Missing host option') } if (opts.log === true) { diff --git a/lib/Serializer.js b/lib/Serializer.js index 97516a2c0..de3e70721 100644 --- a/lib/Serializer.js +++ b/lib/Serializer.js @@ -32,13 +32,18 @@ class Serializer { } var ndjson = '' for (var i = 0, len = array.length; i < len; i++) { - ndjson += this.serialize(array[i]) + '\n' + if (typeof array[i] === 'string') { + ndjson += array[i] + '\n' + } else { + ndjson += this.serialize(array[i]) + '\n' + } } return ndjson } qserialize (object) { debug('qserialize', object) + if (object == null) return '' // arrays should be serialized as comma separated list const keys = Object.keys(object) for (var i = 0, len = keys.length; i < len; i++) { diff --git a/lib/Transport.js b/lib/Transport.js index 420e65767..58e697008 100644 --- a/lib/Transport.js +++ b/lib/Transport.js @@ -2,6 +2,7 @@ const debug = require('debug')('elasticsearch') const once = require('once') +const ms = require('ms') const { ConnectionError, TimeoutError, @@ -18,7 +19,7 @@ class Transport { this.connectionPool = opts.connectionPool this.serializer = opts.serializer this.maxRetries = opts.maxRetries - this.requestTimeout = opts.requestTimeout + this.requestTimeout = toMs(opts.requestTimeout) this.sniffInterval = opts.sniffInterval this.sniffOnConnectionFault = opts.sniffOnConnectionFault this.sniffEndpoint = opts.sniffEndpoint @@ -34,10 +35,11 @@ class Transport { request (params, callback) { callback = once(callback) + const result = { body: null, statusCode: null, headers: null } const attempts = params[kRemainingAttempts] || params.maxRetries || this.maxRetries const connection = this.getConnection() if (connection === null) { - return callback(new NoLivingConnectionsError('There are not living connections')) + return callback(new NoLivingConnectionsError('There are not living connections'), result) } // handle json body @@ -46,7 +48,7 @@ class Transport { try { params.body = this.serializer.serialize(params.body) } catch (err) { - return callback(err) + return callback(err, result) } } params.headers = params.headers || {} @@ -58,7 +60,7 @@ class Transport { try { params.body = this.serializer.ndserialize(params.bulkBody) } catch (err) { - return callback(err) + return callback(err, result) } } else { params.body = params.bulkBody @@ -70,7 +72,9 @@ class Transport { // serializes the querystring params.querystring = this.serializer.qserialize(params.querystring) - params.timeout = params.timeout || this.requestTimeout + // handles request timeout + params.timeout = toMs(params.requestTimeout || this.requestTimeout) + this.emit('request', params) const request = connection.request(params, (err, response) => { @@ -91,30 +95,33 @@ class Transport { : new ConnectionError(err.message, params) this.emit('error', error, params) - return callback(error) + return callback(error, result) } - var json = '' - response.setEncoding('utf8') - response.on('data', chunk => { json += chunk }) - response.on('error', err => callback(new ConnectionError(err.message, params))) - response.on('end', () => { - debug('JSON response', params, json) + const { statusCode, headers } = response + result.statusCode = statusCode + result.headers = headers - const contentType = response.headers['content-type'] - if (contentType != null && contentType.indexOf('application/json') > -1) { + var payload = '' + response.setEncoding('utf8') + response.on('data', chunk => { payload += chunk }) + response.on('error', err => callback(new ConnectionError(err.message, params), result)) + response.on('end', () => { + const isHead = params.method === 'HEAD' + const shouldDeserialize = headers['content-type'] != null && isHead === false && payload !== '' + if (shouldDeserialize === true && headers['content-type'].indexOf('application/json') > -1) { try { - var payload = this.serializer.deserialize(json) + result.body = this.serializer.deserialize(payload) } catch (err) { this.emit('error', err) - return callback(err) + return callback(err, result) } } else { - payload = json + result.body = isHead === true ? true : payload } - const { statusCode, headers } = response - const ignoreStatusCode = Array.isArray(params.ignore) && params.ignore.indexOf(statusCode) > -1 + const ignoreStatusCode = (Array.isArray(params.ignore) && params.ignore.indexOf(statusCode) > -1) || + (isHead === true && statusCode === 404) if (ignoreStatusCode === false && (statusCode === 502 || statusCode === 503 || statusCode === 504)) { @@ -128,11 +135,14 @@ class Transport { this.connectionPool.markAlive(connection) } - this.emit('response', params, { statusCode, payload, headers }) + this.emit('response', params, result) if (ignoreStatusCode === false && statusCode >= 400) { - callback(new ResponseError(payload, statusCode, headers)) + callback(new ResponseError(result), result) } else { - callback(null, payload) + if (isHead === true && statusCode === 404) { + result.body = false + } + callback(null, result) } }) }) @@ -184,4 +194,11 @@ class Transport { } } +function toMs (time) { + if (typeof time === 'string') { + return ms(time) + } + return time +} + module.exports = Transport diff --git a/lib/errors.js b/lib/errors.js index 364808168..91a8b0038 100644 --- a/lib/errors.js +++ b/lib/errors.js @@ -1,14 +1,5 @@ 'use strict' -class BadConfigurationError extends Error { - constructor (message) { - super() - Error.captureStackTrace(this, BadConfigurationError) - this.name = 'BadConfigurationError' - this.message = message || 'Bad Configuration Error' - } -} - class TimeoutError extends Error { constructor (message, request) { super() @@ -56,24 +47,35 @@ class DeserializationError extends Error { } } +class ConfigurationError extends Error { + constructor (message) { + super() + Error.captureStackTrace(this, ConfigurationError) + this.name = 'ConfigurationError' + this.message = message || 'Configuration Error' + } +} + class ResponseError extends Error { - constructor (payload, statusCode, headers) { + constructor ({ body, statusCode, headers }) { super() Error.captureStackTrace(this, ResponseError) this.name = 'ResponseError' - this.message = (payload && payload.error && payload.error.type) || 'Response Error' - this.response = payload - this.statusCode = (payload && payload.status) || statusCode + this.message = (body && body.error && body.error.type) || 'Response Error' + this.body = body + this.statusCode = body && typeof body.status === 'number' + ? body.status + : statusCode this.headers = headers } } module.exports = { - BadConfigurationError, TimeoutError, ConnectionError, NoLivingConnectionsError, SerializationError, DeserializationError, + ConfigurationError, ResponseError } diff --git a/package.json b/package.json index 1bce1c6ae..a99fcf5d3 100644 --- a/package.json +++ b/package.json @@ -43,6 +43,7 @@ }, "dependencies": { "debug": "^4.1.0", + "ms": "^2.1.1", "once": "^1.4.0", "simple-get": "^3.0.3" },