habsi/elasticsearch-js
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP: initial prototype

Browse Source 
- 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
This commit is contained in:
delvedor
2018-10-30 16:32:10 +01:00
parent 961b8224ef
commit c9635c4a71
5 changed files with 64 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
index.js
View File

@ -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) {

7
lib/Serializer.js
View File

@ -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++) {

61
lib/Transport.js
View File

@ -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

30
lib/errors.js
View File

@ -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
}

1
package.json
View File

@ -43,6 +43,7 @@
},
"dependencies": {
"debug": "^4.1.0",
"ms": "^2.1.1",
"once": "^1.4.0",
"simple-get": "^3.0.3"
},