Updated test

test/unit/connection-pool.test.js

@@ -5,6 +5,7 @@ const { URL } = require('url')
 const ConnectionPool = require('../../lib/ConnectionPool')
 const Connection = require('../../lib/Connection')
 const { RoundRobinSelector } = require('../../lib/Selectors')
+const { buildServer } = require('../utils')
 
 test('API', t => {
   t.test('addConnection', t => {
@@ -17,6 +18,19 @@ test('API', t => {
     t.end()
   })
 
+  t.test('addConnection should throw with two connections with the same id', t => {
+    const pool = new ConnectionPool({ selector: new RoundRobinSelector() })
+    const href = 'http://localhost:9200/'
+    pool.addConnection(href)
+    try {
+      pool.addConnection(href)
+      t.fail('Should throw')
+    } catch (err) {
+      t.is(err.message, `Connection with id '${href}' is already present`)
+    }
+    t.end()
+  })
+
   t.test('markDead', t => {
     const pool = new ConnectionPool({ selector: new RoundRobinSelector() })
     const href = 'http://localhost:9200/'
@@ -30,6 +44,20 @@ test('API', t => {
     t.end()
   })
 
+  t.test('markDead should sort the dead queue by deadTimeout', t => {
+    const pool = new ConnectionPool({ selector: new RoundRobinSelector() })
+    const href1 = 'http://localhost:9200/1'
+    const href2 = 'http://localhost:9200/2'
+    const conn1 = pool.addConnection(href1)
+    const conn2 = pool.addConnection(href2)
+    pool.markDead(conn2)
+    setTimeout(() => {
+      pool.markDead(conn1)
+      t.deepEqual(pool.dead, [href2, href1])
+      t.end()
+    }, 10)
+  })
+
   t.test('markAlive', t => {
     const pool = new ConnectionPool({ selector: new RoundRobinSelector() })
     const href = 'http://localhost:9200/'
@@ -45,15 +73,98 @@ test('API', t => {
   })
 
   t.test('resurrect', t => {
-    const pool = new ConnectionPool({ selector: new RoundRobinSelector() })
-    const href = 'http://localhost:9200/'
-    var connection = pool.addConnection(href)
-    pool.markDead(connection)
-    connection = pool.resurrect(Date.now() + 1000 * 60 * 3)
-    t.strictEqual(connection.deadCount, 1)
-    t.true(connection.resurrectTimeout > 0)
-    t.deepEqual(pool.alive, [href])
-    t.deepEqual(pool.dead, [])
+    t.test('ping strategy', t => {
+      t.test('alive', t => {
+        function handler (req, res) {
+          res.end()
+        }
+
+        buildServer(handler, ({ port }, server) => {
+          const pool = new ConnectionPool({
+            selector: new RoundRobinSelector(),
+            resurrectStrategy: 'ping',
+            pingTimeout: 3000
+          })
+          const href = `http://localhost:${port}/`
+          var connection = pool.addConnection(href)
+          pool.markDead(connection)
+          pool.resurrect(Date.now() + 1000 * 60 * 3, (isAlive, connection) => {
+            t.true(isAlive)
+            connection = pool.connections.get(connection.id)
+            t.strictEqual(connection.deadCount, 0)
+            t.strictEqual(connection.resurrectTimeout, 0)
+            t.deepEqual(pool.alive, [href])
+            t.deepEqual(pool.dead, [])
+            t.end()
+          })
+        })
+      })
+
+      t.test('dead', t => {
+        buildServer(() => {}, ({ port }, server) => {
+          server.close()
+          const pool = new ConnectionPool({
+            selector: new RoundRobinSelector(),
+            resurrectStrategy: 'ping',
+            pingTimeout: 3000
+          })
+          const href = `http://localhost:${port}/`
+          var connection = pool.addConnection(href)
+          pool.markDead(connection)
+          pool.resurrect(Date.now() + 1000 * 60 * 3, (isAlive, connection) => {
+            t.false(isAlive)
+            connection = pool.connections.get(connection.id)
+            t.strictEqual(connection.deadCount, 2)
+            t.true(connection.resurrectTimeout > 0)
+            t.deepEqual(pool.alive, [])
+            t.deepEqual(pool.dead, [href])
+            t.end()
+          })
+        })
+      })
+
+      t.end()
+    })
+
+    t.test('optimistic strategy', t => {
+      const pool = new ConnectionPool({
+        selector: new RoundRobinSelector(),
+        resurrectStrategy: 'optimistic'
+      })
+      const href = 'http://localhost:9200/'
+      var connection = pool.addConnection(href)
+      pool.markDead(connection)
+      pool.resurrect(Date.now() + 1000 * 60 * 3, (isAlive, connection) => {
+        t.ok(isAlive === null)
+        connection = pool.connections.get(connection.id)
+        t.strictEqual(connection.deadCount, 1)
+        t.true(connection.resurrectTimeout > 0)
+        t.deepEqual(pool.alive, [href])
+        t.deepEqual(pool.dead, [])
+        t.end()
+      })
+    })
+
+    t.test('none strategy', t => {
+      const pool = new ConnectionPool({
+        selector: new RoundRobinSelector(),
+        resurrectStrategy: 'none'
+      })
+      const href = 'http://localhost:9200/'
+      var connection = pool.addConnection(href)
+      pool.markDead(connection)
+      pool.resurrect(Date.now() + 1000 * 60 * 3, (isAlive, connection) => {
+        t.ok(isAlive === null)
+        t.ok(connection === null)
+        connection = pool.connections.get(href)
+        t.strictEqual(connection.deadCount, 1)
+        t.true(connection.resurrectTimeout > 0)
+        t.deepEqual(pool.alive, [])
+        t.deepEqual(pool.dead, [href])
+        t.end()
+      })
+    })
+
     t.end()
   })
 

test/unit/transport.test.js

@@ -8,7 +8,8 @@ const {
   SerializationError,
   DeserializationError,
   TimeoutError,
-  ResponseError
+  ResponseError,
+  ConnectionError
 } = require('../../lib/errors')
 
 const ConnectionPool = require('../../lib/ConnectionPool')
@@ -94,6 +95,39 @@ test('Send POST', t => {
   })
 })
 
+test('Not JSON payload from server', t => {
+  t.plan(2)
+  function handler (req, res) {
+    res.setHeader('Content-Type', 'text/plain')
+    res.end('hello!')
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const pool = new ConnectionPool({
+      selector: new RoundRobinSelector()
+    })
+    pool.addConnection(`http://localhost:${port}`)
+
+    const transport = new Transport({
+      emit: () => {},
+      connectionPool: pool,
+      serializer: new Serializer(),
+      maxRetries: 3,
+      requestTimeout: 30000,
+      sniffInterval: false,
+      sniffOnStart: false
+    })
+
+    transport.request({
+      method: 'GET',
+      path: '/hello'
+    }, (err, body) => {
+      t.error(err)
+      t.strictEqual(body, 'hello!')
+    })
+  })
+})
+
 test('NoLivingConnectionsError', t => {
   t.plan(1)
   const pool = new ConnectionPool({
@@ -223,6 +257,45 @@ test('TimeoutError (should call markDead on the failing connection)', t => {
   })
 })
 
+test('ConnectionError (should call markDead on the failing connection)', t => {
+  t.plan(2)
+
+  class CustomConnectionPool extends ConnectionPool {
+    markDead (connection) {
+      t.strictEqual(connection.id, 'node1')
+      super.markDead(connection)
+    }
+  }
+
+  buildServer(() => {}, ({ port }, server) => {
+    server.close()
+    const pool = new CustomConnectionPool({
+      selector: new RoundRobinSelector()
+    })
+    pool.addConnection({
+      host: new URL(`http://localhost:${port}`),
+      id: 'node1'
+    })
+
+    const transport = new Transport({
+      emit: () => {},
+      connectionPool: pool,
+      serializer: new Serializer(),
+      maxRetries: 0,
+      requestTimeout: 30000,
+      sniffInterval: false,
+      sniffOnStart: false
+    })
+
+    transport.request({
+      method: 'GET',
+      path: '/hello'
+    }, (err, body) => {
+      t.ok(err instanceof ConnectionError)
+    })
+  })
+})
+
 test('Retry mechanism', t => {
   t.plan(2)
 
@@ -469,3 +542,206 @@ test('Override requestTimeout', t => {
     })
   })
 })
+
+test('sniff', t => {
+  t.test('sniffOnStart', t => {
+    t.plan(4)
+
+    class CustomConnectionPool extends ConnectionPool {
+      empty () {
+        t.ok('called')
+        return this
+      }
+
+      addConnection (hosts) {
+        // the first `addConnection` call should work
+        if (typeof hosts === 'string') {
+          return super.addConnection(hosts)
+        }
+        t.true(Array.isArray(hosts))
+      }
+
+      nodesToHost (nodes) {
+        t.ok('called')
+        return []
+      }
+    }
+
+    function handler (req, res) {
+      t.strictEqual(req.url, '/sniff')
+      res.setHeader('Content-Type', 'application/json;utf=8')
+      res.end(JSON.stringify({ hello: 'world' }))
+    }
+
+    buildServer(handler, ({ port }, server) => {
+      const pool = new CustomConnectionPool({
+        selector: new RoundRobinSelector()
+      })
+      pool.addConnection(`http://localhost:${port}`)
+
+      // eslint-disable-next-line
+      new Transport({
+        emit: () => {},
+        connectionPool: pool,
+        serializer: new Serializer(),
+        maxRetries: 3,
+        requestTimeout: 30000,
+        sniffInterval: false,
+        sniffOnStart: true,
+        sniffEndpoint: '/sniff'
+      })
+    })
+  })
+
+  t.test('sniffOnConnectionFault', t => {
+    t.plan(4)
+
+    class CustomConnectionPool extends ConnectionPool {
+      empty () {
+        t.ok('called')
+        return this
+      }
+
+      addConnection (hosts) {
+        // the first `addConnection` call should work
+        if (typeof hosts === 'string') {
+          return super.addConnection(hosts)
+        }
+
+        t.true(Array.isArray(hosts))
+      }
+
+      nodesToHost (nodes) {
+        t.ok('called')
+        return []
+      }
+    }
+
+    function handler (req, res) {
+      if (req.url === '/other/sniff') {
+        res.setHeader('Content-Type', 'application/json;utf=8')
+        res.end(JSON.stringify({ hello: 'world' }))
+      } else {
+        setTimeout(() => res.end(), 1000)
+      }
+    }
+
+    buildServer(handler, ({ port }, server) => {
+      const pool = new CustomConnectionPool({
+        selector: new RoundRobinSelector()
+      })
+      pool.addConnection(`http://localhost:${port}`)
+      pool.addConnection(`http://localhost:${port}/other`)
+
+      const transport = new Transport({
+        emit: () => {},
+        connectionPool: pool,
+        serializer: new Serializer(),
+        maxRetries: 0,
+        requestTimeout: 500,
+        sniffInterval: false,
+        sniffOnConnectionFault: true,
+        sniffEndpoint: '/sniff'
+      })
+
+      transport.request({
+        method: 'GET',
+        path: '/'
+      }, (err, body) => {
+        t.ok(err instanceof TimeoutError)
+      })
+    })
+  })
+
+  t.test('sniffInterval', t => {
+    t.plan(9)
+
+    class CustomConnectionPool extends ConnectionPool {
+      empty () {
+        return this
+      }
+
+      addConnection (hosts) {
+        // the first `addConnection` call should work
+        if (typeof hosts === 'string') {
+          return super.addConnection(hosts)
+        }
+      }
+
+      nodesToHost (nodes) {
+        return []
+      }
+    }
+
+    function handler (req, res) {
+      // this should be called 6 times
+      t.ok('called')
+      res.setHeader('Content-Type', 'application/json;utf=8')
+      res.end(JSON.stringify({ hello: 'world' }))
+    }
+
+    buildServer(handler, ({ port }, server) => {
+      const pool = new CustomConnectionPool({
+        selector: new RoundRobinSelector()
+      })
+      pool.addConnection(`http://localhost:${port}`)
+
+      const transport = new Transport({
+        emit: () => {},
+        connectionPool: pool,
+        serializer: new Serializer(),
+        maxRetries: 3,
+        requestTimeout: 3000,
+        sniffInterval: 1,
+        sniffEndpoint: '/sniff'
+      })
+
+      const params = { method: 'GET', path: '/' }
+      setTimeout(() => {
+        transport.request(params, t.error)
+      }, 100)
+
+      setTimeout(() => {
+        transport.request(params, t.error)
+      }, 200)
+
+      setTimeout(() => {
+        transport.request(params, t.error)
+      }, 300)
+    })
+  })
+
+  t.test('errored', t => {
+    t.plan(1)
+
+    class CustomConnectionPool extends ConnectionPool {
+      nodesToHost () {
+        t.fail('This should not be called')
+      }
+    }
+
+    buildServer(() => {}, ({ port }, server) => {
+      server.close()
+      const pool = new CustomConnectionPool({
+        selector: new RoundRobinSelector()
+      })
+      pool.addConnection(`http://localhost:${port}`)
+
+      const transport = new Transport({
+        emit: () => {},
+        connectionPool: pool,
+        serializer: new Serializer(),
+        maxRetries: 0,
+        requestTimeout: 30000,
+        sniffInterval: false,
+        sniffEndpoint: '/sniff'
+      })
+
+      transport.sniff((err, hosts) => {
+        t.ok(err instanceof ConnectionError)
+      })
+    })
+  })
+
+  t.end()
+})