Accept stream options in constructors, pass to internal transform streams.

Includes tests.

copy-to.js

@@ -1,12 +1,12 @@
-module.exports = function(txt) {
+module.exports = function(txt, options) {
-  return new CopyStreamQuery(txt)
+  return new CopyStreamQuery(txt, options)
 }
 
 var Transform = require('stream').Transform
 var util = require('util')
 
-var CopyStreamQuery = function(text) {
+var CopyStreamQuery = function(text, options) {
-  Transform.call(this)
+  Transform.call(this, options)
   this.text = text
   this._listeners = {}
   this._copyOutResponse = null

index.js

@@ -1,19 +1,19 @@
 var CopyToQueryStream = require('./copy-to')
 
 module.exports = {
-  to: function(txt) {
+  to: function(txt, options) {
-    return new CopyToQueryStream(txt)
+    return new CopyToQueryStream(txt, options)
   },
-  from: function (txt) {
+  from: function (txt, options) {
-    return new CopyStreamQuery(txt)
+    return new CopyStreamQuery(txt, options)
   }
 }
 
 var Transform = require('stream').Transform
 var util = require('util')
 
-var CopyStreamQuery = function(text) {
+var CopyStreamQuery = function(text, options) {
-  Transform.call(this)
+  Transform.call(this, options)
   this.text = text
   this._listeners = null
   this._copyOutResponse = null

test/copy-from.js

@@ -5,20 +5,30 @@ var concat = require('concat-stream')
 var _ = require('lodash')
 var pg = require('pg.js')
 
-var testRange = function(top) {
+var copy = require('../').from
-  var client = function() {
+
-    var client = new pg.Client()
+var client = function() {
-    client.connect()
+  var client = new pg.Client()
-    client.query('CREATE TEMP TABLE numbers(num int, bigger_num int)')
+  client.connect()
-    return client
+  return client
+}
+
+var testConstruction = function() {
+  var highWaterMark = 10
+  var stream = copy('COPY numbers FROM STDIN', {highWaterMark: 10, objectMode: true})
+  for(var i = 0; i < highWaterMark * 1.5; i++) {
+    stream.write('1\t2\n')
   }
+  assert(!stream.write('1\t2\n'), 'Should correctly set highWaterMark.')
+}
 
+testConstruction()
+
+var testRange = function(top) {
   var fromClient = client()
-  var copy = require('../').from
+  fromClient.query('CREATE TEMP TABLE numbers(num int, bigger_num int)')
-
 
   var txt = 'COPY numbers FROM STDIN'
-
   var stream = fromClient.query(copy(txt))
   var rowEmitCount = 0
   stream.on('row', function() {

test/copy-to.js

@@ -14,6 +14,14 @@ var client = function() {
   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'