Bumped v7.0.0-rc.2

Node.js v6 will go EOL at the end of April 2019, and already two of the production dependencies of the client have already dropped support for it, and soon others will do *(as well as development dependencies)*.

Furthermore, since Node.js will go in EOL it will never get security patches, plus, also OpenSSL-1.0.2 will go EOL [this year](https://github.com/nodejs/Release#release-schedule); to avoid risks for the client users it is better to drop support for Node.js v6 right away.

.ci/test-matrix.yml

@@ -1,11 +1,10 @@
 ---
 ELASTICSEARCH_VERSION:
-- 7.0.0-rc1
+- 7.0.0
 
 NODE_JS_VERSION:
 - 11
 - 10
 - 8
-- 6
 
 exclude: ~

.npmignore

@@ -67,3 +67,6 @@ scripts
 .ci
 .travis.yml
 certs
+.github
+CODE_OF_CONDUCT.md
+CONTRIBUTING.md

.travis.yml

@@ -5,13 +5,13 @@ sudo: required
 language: node_js
 
 node_js:
+  - "11"
   - "10"
   - "8"
-  - "6"
 
 env:
   global:
-    - ELASTICSEARCH_VERSION=7.0.0-rc1
+    - ELASTICSEARCH_VERSION=7.0.0
     - QUIET=true
 
 before_install:

README.md

@@ -6,8 +6,7 @@
 
 ---
 
-**Note:** In the past months we have worked on the new Elasticsearch Node.js client, we will announce it soon!
-If you want you can already try it by following the instructions below, while if you're going to use the legacy one or report an issue, please check out [elastic/elasticsearch-js-legacy](https://github.com/elastic/elasticsearch-js-legacy).
+**Note:** In the past months we have worked on the new Elasticsearch Node.js client, and if you want you can already try it by following the instructions below, while if you're going to use the legacy one or report an issue, please check out [elastic/elasticsearch-js-legacy](https://github.com/elastic/elasticsearch-js-legacy).
 
 ---
 
@@ -28,11 +27,7 @@ npm install @elastic/elasticsearch
 
 ### Compatibility
 
----
-
-**Warning:** currently on npm you will only find a beta release that works with Elasticsearch 7, we will add the support for Elasticsearch 5, 6, and 7 once we release the final version of this library.
-
----
+The minimum supported version of Node.js is `v8`.
 
 The library is compatible with all Elasticsearch versions since 5.x, but you should use the same major version of the Elasticsearch instance that you are using.
 ```
@@ -47,7 +42,9 @@ The library is compatible with all Elasticsearch versions since 5.x, but you sho
 ```
 
 ## Usage
-You can find the full documentation in the [docs](https://github.com/elastic/elasticsearch-js/tree/master/docs) folder.
+
+You can find the full documentation in our [docs](https://www.elastic.co/guide/en/elasticsearch/client/javascript-api/current/index.html) website.
+
 ```js
 const { Client } = require('@elastic/elasticsearch')
 const client = new Client({ node: 'http://localhost:9200' })
@@ -77,7 +74,9 @@ The returned value of **every** API call is formed as follows:
 }
 ```
 ### Client options
+
 The client is designed to be easily configured as you see fit for your needs, following you can see all the possible options that you can use to configure it.
+
 ```ts
 {
   // the Elasticsearch endpoint to use

docs/breaking-changes.asciidoc

@@ -9,7 +9,7 @@ Every breaking change was carefully weighed, and each is justified. Furthermore,
 
 === Breaking changes
 
-*  Minimum supported version of Node.js is `v6`.
+*  Minimum supported version of Node.js is `v8`.
 
 * Everything has been rewritten using ES6 classes to help users extend the defaults more easily.
 

docs/configuration.asciidoc

@@ -90,8 +90,20 @@ _Default:_ `false`
 _Default:_ `null`
 
 |`agent`
-|`http.AgentOptions` - http agent https://nodejs.org/api/http.html#http_new_agent_options[options]. +
+a|`http.AgentOptions, function` - http agent https://nodejs.org/api/http.html#http_new_agent_options[options], or a function that returns an actual http agent instance. +
 _Default:_ `null`
+[source,js]
+----
+const client = new Client({
+  node: 'http://localhost:9200',
+  agent: { agent: 'options' }
+})
+
+const client = new Client({
+  node: 'http://localhost:9200',
+  agent: () => new CustomAgent()
+})
+----
 
 |`nodeFilter`
 a|`function` - Filters which node not to use for a request. +

docs/introduction.asciidoc

@@ -18,6 +18,7 @@ npm install @elastic/elasticsearch
 ----
 
 === Compatibility
+The minimum supported version of Node.js is `v8`.
 
 The library is compatible with all Elasticsearch versions since 5.x, but you should use the same major version of the Elasticsearch instance that you are using.
 ----
@@ -50,4 +51,4 @@ client.search({
 }, (err, result) => {
   if (err) console.log(err)
 })
-----
+----

docs/typescript.asciidoc

@@ -3,58 +3,146 @@
 
 The client offers a first-class support for TypeScript, since it ships the type definitions for every exposed API.
 
-While the client offers type definitions for Request parameters, Request bodies and responses are shipped with `any` because there is not an official spec that defines them, so we cannot make guarantees over them (but since they are shipped with `any`, you can easily override them with your own typing definitions).
-
 NOTE: If you are using TypeScript you will be required to use _snake_case_ style to define the API parameters instead of _camelCase_. 
 
-=== How to extend the provided typings?
-Extend the provided typings is very straightforward, you should declare a custom `.d.ts` file and then write inside your type extensions, following there is an example of how do it.
+Other than the types for the surface API, the client offers the types for every request method, via the `RequestParams`, if you need the types for a search request for instance, you can access them via `RequestParams.Search`.
+Every API that supports a body, accepts a https://www.typescriptlang.org/docs/handbook/generics.html[generics] which represents the type of the request body, if you don't configure anything, it will default to `any`.
+
+For example:
+
 [source,ts]
 ----
-declare module '@elastic/elasticsearch' {
-  export interface ShardsResponse {
-    total: number;
-    successful: number;
-    failed: number;
-    skipped: number;
-  }
+import { RequestParams } from '@elastic/elasticsearch'
 
-   export interface Explanation {
-    value: number;
-    description: string;
-    details: Explanation[];
-  }
-
-   export interface SearchResponse<T> {
-    took: number;
-    timed_out: boolean;
-    _scroll_id?: string;
-    _shards: ShardsResponse;
-    hits: {
-      total: number;
-      max_score: number;
-      hits: Array<{
-        _index: string;
-        _type: string;
-        _id: string;
-        _score: number;
-        _source: T;
-        _version?: number;
-        _explanation?: Explanation;
-        fields?: any;
-        highlight?: any;
-        inner_hits?: any;
-        matched_queries?: string[];
-        sort?: string[];
-      }>;
-    };
-    aggregations?: any;
-  }
-
-   export interface MSearchResponse<T> {
-    responses?: Array<SearchResponse<T>>;
+interface SearchBody {
+  query: {
+    match: { foo: string }
   }
 }
 
-export {};
+const searchParams: RequestParams.Search<SearchBody> = {
+  index: 'test',
+  body: {
+    query: {
+      match: { foo: 'bar' }
+    }
+  }
+}
+
+// This is valid as well
+const searchParams: RequestParams.Search = {
+  index: 'test',
+  body: {
+    query: {
+      match: { foo: 'bar' }
+    }
+  }
+}
+----
+
+You can find the type definiton of a response in `ApiResponse`, which accepts a generics as well if you want to specify the body type, otherwise it defaults to `any`.
+
+[source,ts]
+----
+interface SearchResponse<T> {
+  hits: {
+    hits: Array<{
+      _source: T;
+    }>
+  }
+}
+
+// Define the intefrace of the source object
+interface Source {
+  foo: string
+}
+
+client.search(searchParams)
+  .then((response: ApiResponse<SearchResponse<Source>>) => console.log(response))
+  .catch((err: Error) => {})
+----
+
+=== A complete example
+
+[source,ts]
+----
+import {
+  Client,
+  // Object that contains the type definitions of every API method
+  RequestParams,
+  // Interface of the generic API response
+  ApiResponse,
+} from '@elastic/elasticsearch'
+
+const client = new Client({ node: 'http://localhost:9200' })
+
+// Define the type of the body for the Search request
+interface SearchBody {
+  query: {
+    match: { foo: string }
+  }
+}
+
+// Complete definition of the Search response
+interface ShardsResponse {
+  total: number;
+  successful: number;
+  failed: number;
+  skipped: number;
+}
+
+interface Explanation {
+  value: number;
+  description: string;
+  details: Explanation[];
+}
+
+interface SearchResponse<T> {
+  took: number;
+  timed_out: boolean;
+  _scroll_id?: string;
+  _shards: ShardsResponse;
+  hits: {
+    total: number;
+    max_score: number;
+    hits: Array<{
+      _index: string;
+      _type: string;
+      _id: string;
+      _score: number;
+      _source: T;
+      _version?: number;
+      _explanation?: Explanation;
+      fields?: any;
+      highlight?: any;
+      inner_hits?: any;
+      matched_queries?: string[];
+      sort?: string[];
+    }>;
+  };
+  aggregations?: any;
+}
+
+// Define the intefrace of the source object
+interface Source {
+  foo: string
+}
+
+async function run (): Promise<void> {
+  // Define the search parameters
+  const searchParams: RequestParams.Search<SearchBody> = {
+    index: 'test',
+    body: {
+      query: {
+        match: { foo: 'bar' }
+      }
+    }
+  }
+
+  // Craft the final type definition
+  const response: ApiResponse<SearchResponse<Source>> = await client.search(searchParams)
+  console.log(response.body)
+}
+
+run().catch(console.log)
 ----

docs/usage.asciidoc

@@ -66,7 +66,7 @@ When using the callback style API, the function will also return an object that
 [source,js]
 ----
 // calback API
-const requesty = client.search({
+const request = client.search({
   index: 'my-index',
   body: { foo: 'bar' }
 }, {
@@ -302,4 +302,4 @@ export interface ResurrectEvent {
   isAlive: boolean;
   connection: Connection;
 }
-----
+----

index.d.ts

@@ -27,9 +27,10 @@ import Transport, {
   TransportRequestParams,
   TransportRequestOptions,
   nodeFilterFn,
-  nodeSelectorFn
+  nodeSelectorFn,
+  TransportRequestCallback
 } from './lib/Transport';
-import Connection, { AgentOptions } from './lib/Connection';
+import Connection, { AgentOptions, agentFn } from './lib/Connection';
 import ConnectionPool, { ResurrectEvent } from './lib/ConnectionPool';
 import Serializer from './lib/Serializer';
 import * as RequestParams from './api/requestParams';
@@ -38,12 +39,18 @@ import * as errors from './lib/errors';
 declare type anyObject = {
   [key: string]: any;
 };
-declare type callbackFn = (err: Error | null, result: ApiResponse) => void;
 
-interface ApiMethod<T> {
-  (callback?: callbackFn): any;
-  (params: T, callback?: callbackFn): any;
-  (params: T, options: TransportRequestOptions, callback?: callbackFn): any;
+declare type callbackFn<T> = (err: Error | null, result: ApiResponse<T>) => void;
+
+interface ApiMethod<TParams, TBody = any> {
+  // Promise API
+  (): Promise<ApiResponse<TBody>>;
+  (params: TParams): Promise<ApiResponse<TBody>>;
+  (params: TParams, options: TransportRequestOptions): Promise<ApiResponse<TBody>>;
+  // Callback API
+  (callback: callbackFn<TBody>): TransportRequestCallback;
+  (params: TParams, callback: callbackFn<TBody>): TransportRequestCallback;
+  (params: TParams, options: TransportRequestOptions, callback: callbackFn<TBody>): TransportRequestCallback;
 }
 
 // Extend API
@@ -82,7 +89,7 @@ interface ClientOptions {
   suggestCompression?: boolean;
   compression?: 'gzip';
   ssl?: SecureContextOptions;
-  agent?: AgentOptions;
+  agent?: AgentOptions | agentFn;
   nodeFilter?: nodeFilterFn;
   nodeSelector?: nodeSelectorFn | string;
   headers?: anyObject;
@@ -570,5 +577,6 @@ export {
   RequestEvent,
   ResurrectEvent,
   RequestParams,
+  ClientOptions,
   ClientExtendsCallbackOptions
 };

lib/Connection.d.ts

@@ -24,12 +24,14 @@ import { inspect, InspectOptions } from 'util';
 import * as http from 'http';
 import { SecureContextOptions } from 'tls';
 
+export declare type agentFn = () => any;
+
 interface ConnectionOptions {
   url: URL;
   ssl?: SecureContextOptions;
   id?: string;
   headers?: any;
-  agent?: AgentOptions;
+  agent?: AgentOptions | agentFn;
   status?: string;
   roles?: any;
 }

lib/Connection.js

@@ -46,18 +46,20 @@ class Connection {
       throw new ConfigurationError(`Invalid protocol: '${this.url.protocol}'`)
     }
 
-    // Probably there is a bug in Node Core
-    // see https://github.com/nodejs/node/issues/26357
-    const keepAliveFalse = opts.agent && opts.agent.keepAlive === false
-    const agentOptions = Object.assign({}, {
-      keepAlive: true,
-      keepAliveMsecs: 1000,
-      maxSockets: keepAliveFalse ? Infinity : 256,
-      maxFreeSockets: 256
-    }, opts.agent)
-    this.agent = this.url.protocol === 'http:'
-      ? new http.Agent(agentOptions)
-      : new https.Agent(Object.assign({}, agentOptions, this.ssl))
+    if (typeof opts.agent === 'function') {
+      this.agent = opts.agent()
+    } else {
+      const keepAliveFalse = opts.agent && opts.agent.keepAlive === false
+      const agentOptions = Object.assign({}, {
+        keepAlive: true,
+        keepAliveMsecs: 1000,
+        maxSockets: keepAliveFalse ? Infinity : 256,
+        maxFreeSockets: 256
+      }, opts.agent)
+      this.agent = this.url.protocol === 'http:'
+        ? new http.Agent(agentOptions)
+        : new https.Agent(Object.assign({}, agentOptions, this.ssl))
+    }
 
     this.makeRequest = this.url.protocol === 'http:'
       ? http.request

lib/ConnectionPool.d.ts

@@ -127,7 +127,7 @@ export default class ConnectionPool {
    * @param {object} nodes
    * @returns {array} hosts
    */
-  nodesToHost(nodes: any): any[];
+  nodesToHost(nodes: any, protocol: string): any[];
   /**
    * Transforms an url string to a host object
    *

lib/ConnectionPool.js

@@ -332,18 +332,33 @@ class ConnectionPool {
    * @param {object} nodes
    * @returns {array} hosts
    */
-  nodesToHost (nodes) {
+  nodesToHost (nodes, protocol) {
     const ids = Object.keys(nodes)
     const hosts = []
 
     for (var i = 0, len = ids.length; i < len; i++) {
       const node = nodes[ids[i]]
       // If there is no protocol in
-      // the `publish_address` new URL wil throw
+      // the `publish_address` new URL will throw
+      // the publish_address can have two forms:
+      //   - ip:port
+      //   - hostname/ip:port
+      // if we encounter the second case, we should
+      // use the hostname instead of the ip
       var address = node.http.publish_address
+      const hostAndIpRegex = /^[a-z0-9_.-]*\/\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b/gi
+      const match = address.match(hostAndIpRegex)
+      if (match !== null) {
+        const ipRegex = /\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b/
+        const ip = address.match(ipRegex)[0]
+        // extract the hostname, the -1 at the end removes the final /
+        const hostname = address.slice(0, address.indexOf(ip) - 1)
+        const port = address.split(':')[1]
+        address = `${hostname}:${port}`
+      }
       address = address.slice(0, 4) === 'http'
         ? address
-        : 'http://' + address
+        : `${protocol}//${address}`
       const roles = node.roles.reduce((acc, role) => {
         acc[role] = true
         return acc

lib/Transport.d.ts

@@ -49,7 +49,7 @@ interface TransportOptions {
   headers?: anyObject;
 }
 
-export interface RequestEvent {
+export interface RequestEvent<T = any> {
   body: any;
   statusCode: number | null;
   headers: anyObject | null;
@@ -71,7 +71,7 @@ export interface RequestEvent {
 
 // ApiResponse and RequestEvent are the same thing
 // we are doing this for have more clear names
-export interface ApiResponse extends RequestEvent {}
+export interface ApiResponse<T = any> extends RequestEvent<T> {}
 
 declare type anyObject = {
   [key: string]: any;
@@ -96,6 +96,10 @@ export interface TransportRequestOptions {
   warnings?: [string];
 }
 
+export interface TransportRequestCallback {
+  abort: () => void;
+}
+
 export default class Transport {
   static sniffReasons: {
     SNIFF_ON_START: string;
@@ -117,7 +121,8 @@ export default class Transport {
   _nextSniff: number;
   _isSniffing: boolean;
   constructor(opts: TransportOptions);
-  request(params: TransportRequestParams, options: TransportRequestOptions, callback: (err: Error | null, result: ApiResponse) => void): any;
+  request(params: TransportRequestParams, options?: TransportRequestOptions): Promise<ApiResponse>;
+  request(params: TransportRequestParams, options?: TransportRequestOptions, callback?: (err: Error | null, result: ApiResponse) => void): TransportRequestCallback;
   getConnection(): Connection | null;
   sniff(callback?: (...args: any[]) => void): void;
 }

lib/Transport.js

@@ -353,7 +353,8 @@ class Transport {
       }
 
       debug('Sniffing ended successfully', result.body)
-      const hosts = this.connectionPool.nodesToHost(result.body.nodes)
+      const protocol = result.meta.connection.url.protocol || 'http:'
+      const hosts = this.connectionPool.nodesToHost(result.body.nodes, protocol)
       this.connectionPool.update(hosts)
 
       result.meta.sniff = { hosts, reason }

package.json

@@ -4,7 +4,7 @@
   "main": "index.js",
   "types": "index.d.ts",
   "homepage": "http://www.elastic.co/guide/en/elasticsearch/client/javascript-api/current/index.html",
-  "version": "7.0.0-rc.1",
+  "version": "7.0.0-rc.2",
   "keywords": [
     "elasticsearch",
     "elastic",
@@ -36,33 +36,33 @@
     "company": "Elasticsearch BV"
   },
   "devDependencies": {
-    "@types/node": "^10.12.24",
-    "codecov": "^3.2.0",
-    "convert-hrtime": "^2.0.0",
+    "@types/node": "^11.13.4",
+    "codecov": "^3.3.0",
+    "convert-hrtime": "^3.0.0",
     "dedent": "^0.7.0",
-    "deepmerge": "^3.1.0",
+    "deepmerge": "^3.2.0",
     "dezalgo": "^1.0.3",
-    "js-yaml": "^3.12.1",
+    "js-yaml": "^3.13.1",
     "license-checker": "^25.0.1",
     "lolex": "^3.1.0",
     "minimist": "^1.2.0",
-    "ora": "^3.2.0",
+    "ora": "^3.4.0",
     "pretty-hrtime": "^1.0.3",
     "rimraf": "^2.6.3",
-    "semver": "^5.6.0",
-    "simple-git": "^1.107.0",
+    "semver": "^6.0.0",
+    "simple-git": "^1.110.0",
     "simple-statistics": "^7.0.2",
-    "split2": "^3.1.0",
+    "split2": "^3.1.1",
     "standard": "^12.0.1",
     "stoppable": "^1.1.0",
-    "tap": "^12.6.0",
-    "typescript": "^3.3.3",
+    "tap": "^12.6.1",
+    "typescript": "^3.4.3",
     "workq": "^2.1.0"
   },
   "dependencies": {
     "debug": "^4.1.1",
-    "decompress-response": "^4.0.0",
-    "into-stream": "^4.0.0",
+    "decompress-response": "^4.2.0",
+    "into-stream": "^5.1.0",
     "ms": "^2.1.1",
     "once": "^1.4.0",
     "pump": "^3.0.0"
@@ -76,6 +76,6 @@
     "url": "https://github.com/elastic/elasticsearch-js/issues"
   },
   "engines": {
-    "node": ">=6"
+    "node": ">=8"
   }
 }

scripts/es-docker-platinum.sh

@@ -28,5 +28,4 @@ exec docker run \
   -v "$repo$testnodekey:/usr/share/elasticsearch/config/certs/testnode.key" \
   -v "$repo$cacrt:/usr/share/elasticsearch/config/certs/ca.crt" \
   -p 9200:9200 \
-  docker.elastic.co/elasticsearch/elasticsearch:7.0.0-rc1
-  # docker.elastic.co/elasticsearch/elasticsearch:6.6.0
+  docker.elastic.co/elasticsearch/elasticsearch:7.0.0

scripts/es-docker.sh

@@ -9,5 +9,4 @@ exec docker run \
   -p 9200:9200 \
   --network=elastic \
   --name=elasticsearch \
-  docker.elastic.co/elasticsearch/elasticsearch:7.0.0-rc1
-  # docker.elastic.co/elasticsearch/elasticsearch:6.6.0
+  docker.elastic.co/elasticsearch/elasticsearch:7.0.0

scripts/kibana-docker.sh

@@ -5,4 +5,4 @@ exec docker run \
   -e ELASTICSEARCH_URL="http://elasticsearch:9200" \
   -p 5601:5601 \
   --network=elastic \
-  docker.elastic.co/kibana/kibana:7.0.0-beta1
+  docker.elastic.co/kibana/kibana:7.0.0

scripts/utils/generateRequestTypes.js

@@ -74,10 +74,10 @@ export interface Generic {
     }
 
     const code = `
-export interface ${name[0].toUpperCase() + name.slice(1)} extends Generic {
+export interface ${name[0].toUpperCase() + name.slice(1)}${body ? '<T = any>' : ''} extends Generic {
   ${partsArr.map(genLine).join('\n  ')}
   ${paramsArr.map(genLine).join('\n  ')}
-  ${body ? `body${body.required ? '' : '?'}: any;` : ''}
+  ${body ? `body${body.required ? '' : '?'}: T;` : ''}
 }
 `
 

test/behavior/sniff.test.js

@@ -93,6 +93,39 @@ test('Should update the connection pool', t => {
   })
 })
 
+test('Should handle hostnames in publish_address', t => {
+  t.plan(10)
+
+  buildCluster({ hostPublishAddress: true }, ({ nodes, shutdown }) => {
+    const client = new Client({
+      node: nodes[Object.keys(nodes)[0]].url
+    })
+    t.strictEqual(client.connectionPool.connections.size, 1)
+
+    client.on(events.SNIFF, (err, request) => {
+      t.error(err)
+      t.strictEqual(
+        request.meta.sniff.reason,
+        Transport.sniffReasons.DEFAULT
+      )
+    })
+
+    // run the sniffer
+    client.transport.sniff((err, hosts) => {
+      t.error(err)
+      t.strictEqual(hosts.length, 4)
+
+      for (var i = 0; i < hosts.length; i++) {
+        // the first node will be an update of the existing one
+        t.strictEqual(hosts[i].url.hostname, 'localhost')
+      }
+
+      t.strictEqual(client.connectionPool.connections.size, 4)
+    })
+    t.teardown(shutdown)
+  })
+})
+
 test('Sniff interval', t => {
   t.plan(10)
 

test/types/index.ts

@@ -22,6 +22,7 @@
 import {
   Client,
   ApiResponse,
+  RequestParams,
   RequestEvent,
   ResurrectEvent,
   events,
@@ -94,6 +95,40 @@ client.index({
   .then((result: ApiResponse) => {})
   .catch((err: Error) => {})
 
+// --- Use generics ---
+// Define the search parameters
+interface SearchBody {
+  query: {
+    match: { foo: string }
+  }
+}
+const searchParams: RequestParams.Search<SearchBody> = {
+  index: 'test',
+  body: {
+    query: {
+      match: { foo: 'bar' }
+    }
+  }
+}
+
+// Dewfine the interface of the search response
+interface SearchResponse<T> {
+  hits: {
+    hits: Array<{
+      _source: T;
+    }>
+  }
+}
+
+// Define the intefrace of the source object
+interface Source {
+  foo: string
+}
+
+client.search(searchParams)
+  .then((response: ApiResponse<SearchResponse<Source>>) => console.log(response))
+  .catch((err: Error) => {})
+
 // extend client
 client.extend('namespace.method', (options: ClientExtendsCallbackOptions) => {
   return function (params: any) {

test/unit/connection-pool.test.js

@@ -282,41 +282,112 @@ test('API', t => {
   })
 
   t.test('nodesToHost', t => {
-    const pool = new ConnectionPool({ Connection })
-    const nodes = {
-      a1: {
-        http: {
-          publish_address: '127.0.0.1:9200'
+    t.test('publish_address as ip address', t => {
+      const pool = new ConnectionPool({ Connection })
+      const nodes = {
+        a1: {
+          http: {
+            publish_address: '127.0.0.1:9200'
+          },
+          roles: ['master', 'data', 'ingest']
         },
-        roles: ['master', 'data', 'ingest']
-      },
-      a2: {
-        http: {
-          publish_address: '127.0.0.1:9202'
-        },
-        roles: ['master', 'data', 'ingest']
+        a2: {
+          http: {
+            publish_address: '127.0.0.1:9201'
+          },
+          roles: ['master', 'data', 'ingest']
+        }
       }
-    }
 
-    t.deepEqual(pool.nodesToHost(nodes), [{
-      url: new URL('http://127.0.0.1:9200'),
-      id: 'a1',
-      roles: {
-        master: true,
-        data: true,
-        ingest: true,
-        ml: false
+      t.deepEqual(pool.nodesToHost(nodes, 'http:'), [{
+        url: new URL('http://127.0.0.1:9200'),
+        id: 'a1',
+        roles: {
+          master: true,
+          data: true,
+          ingest: true,
+          ml: false
+        }
+      }, {
+        url: new URL('http://127.0.0.1:9201'),
+        id: 'a2',
+        roles: {
+          master: true,
+          data: true,
+          ingest: true,
+          ml: false
+        }
+      }])
+
+      t.strictEqual(pool.nodesToHost(nodes, 'http:')[0].url.host, '127.0.0.1:9200')
+      t.strictEqual(pool.nodesToHost(nodes, 'http:')[1].url.host, '127.0.0.1:9201')
+      t.end()
+    })
+
+    t.test('publish_address as host/ip', t => {
+      const pool = new ConnectionPool({ Connection })
+      const nodes = {
+        a1: {
+          http: {
+            publish_address: 'example.com/127.0.0.1:9200'
+          },
+          roles: ['master', 'data', 'ingest']
+        },
+        a2: {
+          http: {
+            publish_address: 'example.com/127.0.0.1:9201'
+          },
+          roles: ['master', 'data', 'ingest']
+        }
       }
-    }, {
-      url: new URL('http://127.0.0.1:9201'),
-      id: 'a2',
-      roles: {
-        master: true,
-        data: true,
-        ingest: true,
-        ml: false
+
+      t.deepEqual(pool.nodesToHost(nodes, 'http:'), [{
+        url: new URL('http://example.com:9200'),
+        id: 'a1',
+        roles: {
+          master: true,
+          data: true,
+          ingest: true,
+          ml: false
+        }
+      }, {
+        url: new URL('http://example.com:9201'),
+        id: 'a2',
+        roles: {
+          master: true,
+          data: true,
+          ingest: true,
+          ml: false
+        }
+      }])
+
+      t.strictEqual(pool.nodesToHost(nodes, 'http:')[0].url.host, 'example.com:9200')
+      t.strictEqual(pool.nodesToHost(nodes, 'http:')[1].url.host, 'example.com:9201')
+      t.end()
+    })
+
+    t.test('Should use the configure protocol', t => {
+      const pool = new ConnectionPool({ Connection })
+      const nodes = {
+        a1: {
+          http: {
+            publish_address: 'example.com/127.0.0.1:9200'
+          },
+          roles: ['master', 'data', 'ingest']
+        },
+        a2: {
+          http: {
+            publish_address: 'example.com/127.0.0.1:9201'
+          },
+          roles: ['master', 'data', 'ingest']
+        }
       }
-    }])
+
+      t.strictEqual(pool.nodesToHost(nodes, 'https:')[0].url.protocol, 'https:')
+      t.strictEqual(pool.nodesToHost(nodes, 'http:')[1].url.protocol, 'http:')
+      t.end()
+    })
+
     t.end()
   })
 

test/unit/connection.test.js

@@ -23,6 +23,7 @@ const { test } = require('tap')
 const { inspect } = require('util')
 const { createGzip, createDeflate } = require('zlib')
 const { URL } = require('url')
+const { Agent } = require('http')
 const intoStream = require('into-stream')
 const { buildServer } = require('../utils')
 const Connection = require('../../lib/Connection')
@@ -149,6 +150,55 @@ test('Basic (https with ssl agent)', t => {
   })
 })
 
+test('Custom http agent', t => {
+  t.plan(5)
+
+  function handler (req, res) {
+    t.match(req.headers, {
+      'x-custom-test': 'true',
+      connection: 'keep-alive'
+    })
+    res.end('ok')
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const agent = new Agent({
+      keepAlive: true,
+      keepAliveMsecs: 1000,
+      maxSockets: 256,
+      maxFreeSockets: 256
+    })
+    agent.custom = true
+    const connection = new Connection({
+      url: new URL(`http://localhost:${port}`),
+      agent: () => agent
+    })
+    t.true(connection.agent.custom)
+    connection.request({
+      path: '/hello',
+      method: 'GET',
+      headers: {
+        'X-Custom-Test': true
+      }
+    }, (err, res) => {
+      t.error(err)
+
+      t.match(res.headers, {
+        connection: 'keep-alive'
+      })
+
+      var payload = ''
+      res.setEncoding('utf8')
+      res.on('data', chunk => { payload += chunk })
+      res.on('error', err => t.fail(err))
+      res.on('end', () => {
+        t.strictEqual(payload, 'ok')
+        server.stop()
+      })
+    })
+  })
+})
+
 test('Disable keep alive', t => {
   t.plan(3)
 

test/utils/buildCluster.js

@@ -53,12 +53,14 @@ function buildCluster (options, callback) {
 
     buildServer(options.handler || handler, ({ port }, server) => {
       nodes[opts.id] = {
-        url: `http://localhost:${port}`,
+        url: `http://127.0.0.1:${port}`,
         server
       }
       sniffResult.nodes[opts.id] = {
         http: {
-          publish_address: `http://localhost:${port}`
+          publish_address: options.hostPublishAddress
+            ? `localhost/127.0.0.1:${port}`
+            : `127.0.0.1:${port}`
         },
         roles: ['master', 'data', 'ingest']
       }