Bump version

Adds a smoke test for copying with BINARY.

README.md

@@ -0,0 +1,93 @@
+## pg-copy-streams
+
+COPY FROM / COPY TO for node-postgres.  Stream from one database to another, and stuff.
+
+## how? what? huh?
+
+Did you know the _all powerful_ PostgreSQL supports streaming binary data directly into and out of a table?
+This means you can take your favorite CSV or TSV or whatever format file and pipe it directly into an existing PostgreSQL table.
+You can also take a table and pipe it directly to a file, another database, stdout, even to `/dev/null` if you're crazy!
+
+What this module gives you is a [Readable](http://nodejs.org/api/stream.html#stream_class_stream_readable) or [Writable](http://nodejs.org/api/stream.html#stream_class_stream_writable) stream directly into/out of a table in your database.
+This mode of interfacing with your table is _very fast_ and _very brittle_.  You are responsible for properly encoding and ordering all your columns. If anything is out of place PostgreSQL will send you back an error.  The stream works within a transaction so you wont leave things in a 1/2 borked state, but it's still good to be aware of.
+
+If you're not familiar with the feature (I wasn't either) you can read this for some good helps: http://www.postgresql.org/docs/9.3/static/sql-copy.html
+
+## examples
+
+### pipe from a table to stdout
+
+```js
+var pg = require('pg');
+var copyTo = require('pg-copy-streams').to;
+
+pg.connect(function(err, client, done) {
+  var stream = client.query(copyTo('COPY my_table TO STDOUT'));
+  stream.pipe(process.stdout);
+  stream.on('end', done);
+  stream.on('error', done);
+});
+```
+
+### pipe from a file to table
+
+```js
+var fs = require('fs');
+var pg = require('pg');
+var copyFrom = require('pg-copy-streams').from;
+
+pg.connect(function(err, client, done) {
+  var stream = client.query(copyFrom('COPY my_table FROM STDIN'));
+  var fileStream = fs.createReadStream('some_file.tdv')
+  fileStream.pipe(stream);
+  fileStream.on('end', done);
+  fileStream.on('error', done);
+});
+```
+
+## install
+
+```sh
+$ npm install pg-copy-streams
+```
+
+## notice
+
+Before you set out on this magical piping journey, you _really_ should read this: http://www.postgresql.org/docs/9.3/static/sql-copy.html, and you might want to take a look at the [tests](https://github.com/brianc/node-pg-copy-streams/tree/master/test) to get an idea of how things work.
+
+## contributing 
+
+Instead of adding a bunch more code to the already bloated [node-postgres](https://github.com/brianc/node-postgres) I am trying to make the internals extensible and work on adding edge-case features as 3rd party modules.
+This is one of those.
+
+Please, if you have any issues with this, open an issue.
+
+Better yet, submit a pull request.  I _love_ pull requests.
+
+Generally how I work is if you submit a few pull requests and you're interested I'll make you a contributor and give you full access to everything.
+
+Since this isn't a module with tons of installs and dependent modules I hope we can work together on this to iterate faster here and make something really useful.
+
+## license
+
+The MIT License (MIT)
+
+Copyright (c) 2013 Brian M. Carlson
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

copy-to.js

@@ -8,18 +8,23 @@ var util = require('util')
 var CopyStreamQuery = function(text) {
   Transform.call(this)
   this.text = text
-  this._listeners = null
+  this._listeners = {}
   this._copyOutResponse = null
   this.rowCount = 0
 }
 
 util.inherits(CopyStreamQuery, Transform)
 
+var eventTypes = ['close', 'data', 'end', 'error']
+
 CopyStreamQuery.prototype.submit = function(connection) {
   connection.query(this.text)
   this.connection = connection
-  this._listeners = connection.stream.listeners('data')
-  connection.stream.removeAllListeners('data')
+  var self = this
+  eventTypes.forEach(function(type) {
+    self._listeners[type] = connection.stream.listeners(type)
+    connection.stream.removeAllListeners(type)
+  })
   connection.stream.pipe(this)
 }
 
@@ -32,10 +37,12 @@ var code = {
 
 CopyStreamQuery.prototype._detach = function() {
   this.connection.stream.unpipe()
-  this.connection.stream.removeAllListeners('data')
   var self = this
-  this._listeners.forEach(function(listener) {
-    self.connection.stream.on('data', listener)
+  eventTypes.forEach(function(type) {
+    self.connection.stream.removeAllListeners(type)
+    self._listeners[type].forEach(function(listener) {
+      self.connection.stream.on(type, listener)
+    })
   })
 }
 

index.js

@@ -55,7 +55,7 @@ CopyStreamQuery.prototype.handleError = function(e) {
   this.emit('error', e)
 }
 
-CopyStreamQuery.prototype.streamData = function(connection) {
+CopyStreamQuery.prototype.handleCopyInResponse = function(connection) {
   this.pipe(connection.stream)
 }
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pg-copy-streams",
-  "version": "0.2.0",
+  "version": "0.2.3",
   "description": "Low-Level COPY TO and COPY FROM streams for PostgreSQL in JavaScript using",
   "main": "index.js",
   "scripts": {
@@ -27,6 +27,7 @@
     "concat-stream": "~1.1.0",
     "gonna": "0.0.0",
     "lodash": "~2.2.1",
-    "heroku-env": "~0.1.1"
+    "heroku-env": "~0.1.1",
+    "async": "~0.2.10"
   }
 }

test/binary.js

@@ -0,0 +1,65 @@
+var assert = require('assert')
+var gonna = require('gonna')
+
+var async = require('async')
+var concat = require('concat-stream')
+var _ = require('lodash')
+var pg = require('pg.js')
+
+var from = require('../').from
+var to = require('../').to
+
+var testBinaryCopy = function() {
+  var client = function() {
+    var client = new pg.Client()
+    client.connect()
+    return client
+  }
+
+  var fromClient = client()
+  var toClient = client()
+
+  queries = [
+    'CREATE TABLE data (num BIGINT, word TEXT)',
+    'INSERT INTO data (num, word) VALUES (1, \'hello\'), (2, \'other thing\'), (3, \'goodbye\')',
+    'CREATE TABLE data_copy (LIKE data INCLUDING ALL)'
+  ]
+
+  async.eachSeries(queries, _.bind(fromClient.query, fromClient), function(err) {
+    assert.ifError(err)
+
+    var fromStream = fromClient.query(to('COPY (SELECT * FROM data) TO STDOUT BINARY'))
+    var toStream = toClient.query(from('COPY data_copy FROM STDIN BINARY'))
+
+    runStream = function(callback) {
+      fromStream.on('error', callback)
+      toStream.on('error', callback)
+      toStream.on('finish', callback)
+      fromStream.pipe(toStream)
+    }
+    runStream(function(err) {
+      assert.ifError(err)
+
+      toClient.query('SELECT * FROM data_copy ORDER BY num', function(err, res){
+        assert.equal(res.rowCount, 3, 'expected 3 rows but got ' + res.rowCount)
+        assert.equal(res.rows[0].num, 1)
+        assert.equal(res.rows[0].word, 'hello')
+        assert.equal(res.rows[1].num, 2)
+        assert.equal(res.rows[1].word, 'other thing')
+        assert.equal(res.rows[2].num, 3)
+        assert.equal(res.rows[2].word, 'goodbye')
+        queries = [
+          'DROP TABLE data',
+          'DROP TABLE data_copy'
+        ]
+        async.each(queries, _.bind(fromClient.query, fromClient), function(err) {
+          assert.ifError(err)
+          fromClient.end()
+          toClient.end()
+        })
+      })
+    })
+  })
+}
+
+testBinaryCopy()

test/copy-to.js

@@ -1,20 +1,21 @@
 var assert = require('assert')
 var gonna = require('gonna')
 
-var concat = require('concat-stream')
 var _ = require('lodash')
+var async = require('async')
+var concat = require('concat-stream')
 var pg = require('pg.js')
 
+var copy = require('../').to
+
+var client = function() {
+  var client = new pg.Client()
+  client.connect()
+  return client
+}
+
 var testRange = function(top) {
-  var client = function() {
-    var client = new pg.Client()
-    client.connect()
-    return client
-  }
-
   var fromClient = client()
-  var copy = require('../').to
-
   var txt = 'COPY (SELECT * from generate_series(0, ' + (top - 1) + ')) TO STDOUT'
 
   var stream = fromClient.query(copy(txt))
@@ -32,3 +33,28 @@ var testRange = function(top) {
 }
 
 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)

test/index.js

@@ -1,2 +1,3 @@
 require('./copy-from')
 require('./copy-to')
+require('./binary')