c4a0e6dd58
* Eliminate detach/reattach strategy as it isn't able to differentiate between on/once and inconsistenly loses unshifted data depending on node version. Instead just split stream and send it to copy stream and the connection.stream at the same time. Disconnecting copy stream just means unpiping. Added handleCopyData to fulfill query contract but ignore the incoming data. Add node 4.2.2 to Travis Minimum postgres 9.2 to allow tests to complete in Travis Remove test that is no longer needed since we no longer disconnect/reconnect listeners * Add resume * Remove node 0.10 and add 0.12 * Re-enable old tests * Add more versions to the travis test matrix
78 lines
2.1 KiB
JavaScript
78 lines
2.1 KiB
JavaScript
var assert = require('assert')
|
|
var gonna = require('gonna')
|
|
|
|
var _ = require('lodash')
|
|
var async = require('async')
|
|
var concat = require('concat-stream')
|
|
var pg = require('pg')
|
|
|
|
var copy = require('../').to
|
|
|
|
var client = function() {
|
|
var client = new pg.Client()
|
|
client.connect()
|
|
return client
|
|
}
|
|
|
|
var testConstruction = function() {
|
|
var txt = 'COPY (SELECT * FROM generate_series(0, 10)) TO STDOUT'
|
|
var stream = copy(txt, {highWaterMark: 10})
|
|
assert.equal(stream._readableState.highWaterMark, 10, 'Client should have been set with a correct highWaterMark.')
|
|
}
|
|
|
|
testConstruction()
|
|
|
|
var testRange = function(top) {
|
|
var fromClient = client()
|
|
var txt = 'COPY (SELECT * from generate_series(0, ' + (top - 1) + ')) TO STDOUT'
|
|
var res;
|
|
|
|
|
|
var stream = fromClient.query(copy(txt))
|
|
var done = gonna('finish piping out', 1000, function() {
|
|
fromClient.end()
|
|
})
|
|
|
|
stream.pipe(concat(function(buf) {
|
|
res = buf.toString('utf8')
|
|
}))
|
|
|
|
stream.on('end', function() {
|
|
var expected = _.range(0, top).join('\n') + '\n'
|
|
assert.equal(res, expected)
|
|
assert.equal(stream.rowCount, top, 'should have rowCount ' + top + ' but got ' + stream.rowCount)
|
|
done()
|
|
});
|
|
}
|
|
|
|
testRange(10000)
|
|
|
|
var testInternalPostgresError = function() {
|
|
var cancelClient = client()
|
|
var queryClient = client()
|
|
|
|
var runStream = function(callback) {
|
|
var txt = "COPY (SELECT pg_sleep(10)) TO STDOUT"
|
|
var stream = queryClient.query(copy(txt))
|
|
stream.on('data', function(data) {
|
|
// Just throw away the data.
|
|
})
|
|
stream.on('error', callback)
|
|
|
|
setTimeout(function() {
|
|
var cancelQuery = "SELECT pg_cancel_backend(pid) FROM pg_stat_activity WHERE query ~ 'pg_sleep' AND NOT query ~ 'pg_cancel_backend'"
|
|
cancelClient.query(cancelQuery)
|
|
}, 50)
|
|
}
|
|
|
|
runStream(function(err) {
|
|
assert.notEqual(err, null)
|
|
var expectedMessage = 'canceling statement due to user request'
|
|
assert.notEqual(err.toString().indexOf(expectedMessage), -1, 'Error message should mention reason for query failure.')
|
|
cancelClient.end()
|
|
queryClient.end()
|
|
})
|
|
}
|
|
|
|
testInternalPostgresError()
|