Bump version

* Expose row count after COPY FROM command

* correct conditional

.travis.yml

@@ -1,7 +1,16 @@
 language: node_js
 node_js:
-  - "0.10"
-  - "0.11"
+  - "0.12"
+  - "4"
+  - "5"
+  - "6"
+
+addons:
+  postgresql: "9.2"
+
+services:
+  - postgresql
+
 before_install:
   - npm install npm --global
 env:

copy-to.js

@@ -8,7 +8,6 @@ var util = require('util')
 var CopyStreamQuery = function(text, options) {
   Transform.call(this, options)
   this.text = text
-  this._listeners = {}
   this._copyOutResponse = null
   this.rowCount = 0
 }
@@ -20,11 +19,7 @@ var eventTypes = ['close', 'data', 'end', 'error']
 CopyStreamQuery.prototype.submit = function(connection) {
   connection.query(this.text)
   this.connection = connection
-  var self = this
-  eventTypes.forEach(function(type) {
-    self._listeners[type] = connection.stream.listeners(type)
-    connection.stream.removeAllListeners(type)
-  })
+  this.connection.removeAllListeners('copyData')
   connection.stream.pipe(this)
 }
 
@@ -36,16 +31,12 @@ var code = {
 }
 
 CopyStreamQuery.prototype._detach = function() {
-  this.connection.stream.unpipe()
-  var self = this
-  eventTypes.forEach(function(type) {
-    self.connection.stream.removeAllListeners(type)
-    self._listeners[type].forEach(function(listener) {
-      self.connection.stream.on(type, listener)
-    })
-  })
+  this.connection.stream.unpipe(this)
+  // Unpipe can drop us out of flowing mode
+  this.connection.stream.resume()
 }
 
+
 CopyStreamQuery.prototype._transform = function(chunk, enc, cb) {
   var offset = 0
   if(this._remainder && chunk) {
@@ -55,7 +46,6 @@ CopyStreamQuery.prototype._transform = function(chunk, enc, cb) {
     this._copyOutResponse = true
     if(chunk[0] == code.E) {
       this._detach()
-      this.connection.stream.unshift(chunk)
       this.push(null)
       return cb();
     }
@@ -71,11 +61,6 @@ CopyStreamQuery.prototype._transform = function(chunk, enc, cb) {
     //complete or error
     if(messageCode == code.c || messageCode == code.E) {
       this._detach()
-      if (messageCode == code.c) {
-        this.connection.stream.unshift(chunk.slice(offset + 5))
-      } else {
-        this.connection.stream.unshift(chunk.slice(offset))
-      }
       this.push(null)
       return cb();
     }
@@ -108,6 +93,9 @@ CopyStreamQuery.prototype.handleError = function(e) {
   this.emit('error', e)
 }
 
+CopyStreamQuery.prototype.handleCopyData = function(chunk) {
+}
+
 CopyStreamQuery.prototype.handleCommandComplete = function() {
 }
 

index.js

@@ -1,5 +1,4 @@
 var CopyToQueryStream = require('./copy-to')
-
 module.exports = {
   to: function(txt, options) {
     return new CopyToQueryStream(txt, options)
@@ -40,7 +39,6 @@ CopyStreamQuery.prototype._transform = function(chunk, enc, cb) {
   lenBuffer.writeUInt32BE(chunk.length + 4, 0)
   this.push(lenBuffer)
   this.push(chunk)
-  this.rowCount++
   cb()
 }
 
@@ -59,7 +57,14 @@ CopyStreamQuery.prototype.handleCopyInResponse = function(connection) {
   this.pipe(connection.stream)
 }
 
-CopyStreamQuery.prototype.handleCommandComplete = function() {
+CopyStreamQuery.prototype.handleCommandComplete = function(msg) {
+  // Parse affected row count as in
+  // https://github.com/brianc/node-postgres/blob/35e5567f86774f808c2a8518dd312b8aa3586693/lib/result.js#L37
+  var match = /COPY (\d+)/.exec((msg || {}).text)
+  if (match) {
+    this.rowCount = parseInt(match[1], 10)
+  }
+
   this.unpipe()
   this.emit('end')
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pg-copy-streams",
-  "version": "0.3.0",
+  "version": "1.1.0",
   "description": "Low-Level COPY TO and COPY FROM streams for PostgreSQL in JavaScript using",
   "main": "index.js",
   "scripts": {
@@ -23,7 +23,7 @@
     "url": "https://github.com/brianc/node-pg-copy-streams/issues"
   },
   "devDependencies": {
-    "pg.js": "~2.8.1",
+    "pg": "~4.4.3",
     "concat-stream": "~1.1.0",
     "gonna": "0.0.0",
     "lodash": "~2.2.1",

test/binary.js

@@ -4,7 +4,7 @@ var gonna = require('gonna')
 var async = require('async')
 var concat = require('concat-stream')
 var _ = require('lodash')
-var pg = require('pg.js')
+var pg = require('pg')
 
 var from = require('../').from
 var to = require('../').to

test/copy-from.js

@@ -3,7 +3,7 @@ var gonna = require('gonna')
 
 var concat = require('concat-stream')
 var _ = require('lodash')
-var pg = require('pg.js')
+var pg = require('pg')
 
 var copy = require('../').from
 
@@ -30,10 +30,6 @@ var testRange = function(top) {
 
   var txt = 'COPY numbers FROM STDIN'
   var stream = fromClient.query(copy(txt))
-  var rowEmitCount = 0
-  stream.on('row', function() {
-    rowEmitCount++
-  })
   for(var i = 0; i < top; i++) {
     stream.write(Buffer('' + i + '\t' + i*10 + '\n'))
   }

test/copy-to.js

@@ -4,7 +4,7 @@ var gonna = require('gonna')
 var _ = require('lodash')
 var async = require('async')
 var concat = require('concat-stream')
-var pg = require('pg.js')
+var pg = require('pg')
 
 var copy = require('../').to
 
@@ -25,6 +25,8 @@ 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() {
@@ -32,41 +34,19 @@ var testRange = function(top) {
   })
 
   stream.pipe(concat(function(buf) {
-    var res = buf.toString('utf8')
+    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 testLeak = function(rounds) {
-  var fromClient = client()
-  var txt = 'COPY (SELECT 10) TO STDOUT'
-
-  var runStream = function(num, callback) {
-    var stream = fromClient.query(copy(txt))
-    stream.on('data', function(data) {
-      // Just throw away the data.
-    })
-    stream.on('end', callback)
-    stream.on('error', callback)
-  }
-
-  async.timesSeries(rounds, runStream, function(err) {
-    assert.equal(err, null)
-    assert.equal(fromClient.connection.stream.listeners('close').length, 0)
-    assert.equal(fromClient.connection.stream.listeners('data').length, 1)
-    assert.equal(fromClient.connection.stream.listeners('end').length, 2)
-    assert.equal(fromClient.connection.stream.listeners('error').length, 1)
-    fromClient.end()
-  })
-}
-
-testLeak(5)
-
 var testInternalPostgresError = function() {
   var cancelClient = client()
   var queryClient = client()