committed by
GitHub
parent
203bab278d
commit
fd8f02b9df
@ -49,8 +49,7 @@ test('Should execute the recurrect API with the ping strategy', t => {
|
||||
})
|
||||
|
||||
q.add((q, done) => {
|
||||
cluster.kill('node0')
|
||||
setTimeout(done, 100)
|
||||
cluster.kill('node0', done)
|
||||
})
|
||||
|
||||
q.add((q, done) => {
|
||||
@ -173,8 +172,7 @@ test('Should execute the recurrect API with the optimistic strategy', t => {
|
||||
})
|
||||
|
||||
q.add((q, done) => {
|
||||
cluster.kill('node0')
|
||||
setTimeout(done, 100)
|
||||
cluster.kill('node0', done)
|
||||
})
|
||||
|
||||
q.add((q, done) => {
|
||||
|
||||
@ -6,6 +6,8 @@
|
||||
|
||||
const { test } = require('tap')
|
||||
const { URL } = require('url')
|
||||
const lolex = require('lolex')
|
||||
const workq = require('workq')
|
||||
const { buildCluster } = require('../utils')
|
||||
const { Client, Connection, Transport, events, errors } = require('../../index')
|
||||
|
||||
@ -111,8 +113,10 @@ test('Should handle hostnames in publish_address', t => {
|
||||
})
|
||||
})
|
||||
|
||||
test('Sniff interval', { skip: 'Flaky on CI' }, t => {
|
||||
t.plan(10)
|
||||
test('Sniff interval', t => {
|
||||
t.plan(11)
|
||||
const clock = lolex.install({ toFake: ['Date'] })
|
||||
const q = workq()
|
||||
|
||||
buildCluster(({ nodes, shutdown, kill }) => {
|
||||
const client = new Client({
|
||||
@ -132,19 +136,47 @@ test('Sniff interval', { skip: 'Flaky on CI' }, t => {
|
||||
})
|
||||
|
||||
t.strictEqual(client.connectionPool.size, 1)
|
||||
setTimeout(() => client.info(t.error), 60)
|
||||
|
||||
setTimeout(() => {
|
||||
// let's kill a node
|
||||
kill('node1')
|
||||
client.info(t.error)
|
||||
}, 150)
|
||||
q.add((q, done) => {
|
||||
clock.tick(51)
|
||||
client.info(err => {
|
||||
t.error(err)
|
||||
waitSniffEnd(() => {
|
||||
t.strictEqual(client.connectionPool.size, 4)
|
||||
done()
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
setTimeout(() => {
|
||||
t.strictEqual(client.connectionPool.size, 3)
|
||||
}, 200)
|
||||
q.add((q, done) => {
|
||||
kill('node1', done)
|
||||
})
|
||||
|
||||
q.add((q, done) => {
|
||||
clock.tick(51)
|
||||
client.info(err => {
|
||||
t.error(err)
|
||||
waitSniffEnd(() => {
|
||||
t.strictEqual(client.connectionPool.size, 3)
|
||||
done()
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
t.teardown(shutdown)
|
||||
|
||||
// it can happen that the sniff operation resolves
|
||||
// after the API call that trioggered it, so to
|
||||
// be sure that we are checking the connectionPool size
|
||||
// at the right moment, we verify that the transport
|
||||
// is no longer sniffing
|
||||
function waitSniffEnd (callback) {
|
||||
if (client.transport._isSniffing) {
|
||||
setTimeout(waitSniffEnd, 500, callback)
|
||||
} else {
|
||||
callback()
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@ -578,9 +578,8 @@ test('Retry mechanism', t => {
|
||||
if (count > 0) {
|
||||
res.end(JSON.stringify({ hello: 'world' }))
|
||||
} else {
|
||||
setTimeout(() => {
|
||||
res.end(JSON.stringify({ hello: 'world' }))
|
||||
}, 1000)
|
||||
res.statusCode = 504
|
||||
res.end(JSON.stringify({ error: true }))
|
||||
}
|
||||
count++
|
||||
}
|
||||
@ -603,7 +602,6 @@ test('Retry mechanism', t => {
|
||||
connectionPool: pool,
|
||||
serializer: new Serializer(),
|
||||
maxRetries: 1,
|
||||
requestTimeout: 250,
|
||||
sniffInterval: false,
|
||||
sniffOnStart: false
|
||||
})
|
||||
@ -624,15 +622,10 @@ test('Should not retry if the body is a stream', t => {
|
||||
|
||||
var count = 0
|
||||
function handler (req, res) {
|
||||
res.setHeader('Content-Type', 'application/json;utf=8')
|
||||
if (count > 0) {
|
||||
res.end(JSON.stringify({ hello: 'world' }))
|
||||
} else {
|
||||
setTimeout(() => {
|
||||
res.end(JSON.stringify({ hello: 'world' }))
|
||||
}, 1000)
|
||||
}
|
||||
count++
|
||||
res.setHeader('Content-Type', 'application/json;utf=8')
|
||||
res.statusCode = 504
|
||||
res.end(JSON.stringify({ error: true }))
|
||||
}
|
||||
|
||||
buildServer(handler, ({ port }, server) => {
|
||||
@ -653,7 +646,6 @@ test('Should not retry if the body is a stream', t => {
|
||||
connectionPool: pool,
|
||||
serializer: new Serializer(),
|
||||
maxRetries: 1,
|
||||
requestTimeout: 10,
|
||||
sniffInterval: false,
|
||||
sniffOnStart: false
|
||||
})
|
||||
@ -663,7 +655,7 @@ test('Should not retry if the body is a stream', t => {
|
||||
path: '/hello',
|
||||
body: intoStream(JSON.stringify({ hello: 'world' }))
|
||||
}, (err, { body }) => {
|
||||
t.ok(err instanceof TimeoutError)
|
||||
t.ok(err instanceof ResponseError)
|
||||
t.strictEqual(count, 1)
|
||||
server.stop()
|
||||
})
|
||||
@ -679,9 +671,8 @@ test('Custom retry mechanism', t => {
|
||||
if (count > 0) {
|
||||
res.end(JSON.stringify({ hello: 'world' }))
|
||||
} else {
|
||||
setTimeout(() => {
|
||||
res.end(JSON.stringify({ hello: 'world' }))
|
||||
}, 1000)
|
||||
res.statusCode = 504
|
||||
res.end(JSON.stringify({ error: true }))
|
||||
}
|
||||
count++
|
||||
}
|
||||
@ -704,7 +695,6 @@ test('Custom retry mechanism', t => {
|
||||
connectionPool: pool,
|
||||
serializer: new Serializer(),
|
||||
maxRetries: 0,
|
||||
requestTimeout: 250,
|
||||
sniffInterval: false,
|
||||
sniffOnStart: false
|
||||
})
|
||||
|
||||
@ -56,14 +56,17 @@ function buildCluster (options, callback) {
|
||||
|
||||
function shutdown () {
|
||||
debug(`Shutting down cluster '${clusterId}'`)
|
||||
Object.keys(nodes).forEach(kill)
|
||||
for (const id in nodes) {
|
||||
kill(id)
|
||||
}
|
||||
}
|
||||
|
||||
function kill (id) {
|
||||
function kill (id, callback) {
|
||||
debug(`Shutting down cluster node '${id}' (cluster id: '${clusterId}')`)
|
||||
nodes[id].server.stop()
|
||||
const node = nodes[id]
|
||||
delete nodes[id]
|
||||
delete sniffResult.nodes[id]
|
||||
node.server.stop(callback)
|
||||
}
|
||||
|
||||
function spawn (id, callback) {
|
||||
|
||||
Reference in New Issue
Block a user