Remove a bunch of additional code

Author: brianc

packages/pg-query-stream/README.md

@@ -47,7 +47,7 @@ I'm very open to contribution!  Open a pull request with your code or idea and w
 
 The MIT License (MIT)
 
-Copyright (c) 2013 Brian M. Carlson
+Copyright (c) 2013-2019 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

packages/pg-query-stream/index.js

@@ -8,10 +8,6 @@ class PgQueryStream extends Readable {
     super({ objectMode: true, emitClose: true, autoDestroy: true, highWaterMark: batchSize })
     this.cursor = new Cursor(text, values, config)
 
-    this._reading = false
-    this._callbacks = []
-    this._err = undefined;
-
     // delegate Submittable callbacks to cursor
     this.handleRowDescription = this.cursor.handleRowDescription.bind(this.cursor)
     this.handleDataRow = this.cursor.handleDataRow.bind(this.cursor)
@@ -25,42 +21,16 @@ class PgQueryStream extends Readable {
     this.cursor.submit(connection)
   }
 
-  close(callback) {
-    if (this.destroyed) {
-      if (callback) setImmediate(callback)
-    } else {
-      if (callback) this.once('close', callback)
-      this.destroy()
-    }
-  }
-
-  _close() {
+  _destroy(_err, cb) {
     this.cursor.close((err) => {
-      let cb
-      while ((cb = this._callbacks.pop())) cb(err || this._err)
+      cb && cb(err || _err)
     })
   }
 
-  _destroy(_err, callback) {
-    this._err = _err;
-    this._callbacks.push(callback)
-    if (!this._reading) {
-      this._close()
-    }
-  }
-
   // https://nodejs.org/api/stream.html#stream_readable_read_size_1
   _read(size) {
-    // Prevent _destroy() from closing while reading
-    this._reading = true
-
     this.cursor.read(size, (err, rows, result) => {
-      this._reading = false
-
-      if (this.destroyed) {
-        // Destroyed while reading?
-        this._close()
-      } else if (err) {
+      if (err) {
         // https://nodejs.org/api/stream.html#stream_errors_while_reading
         this.destroy(err)
       } else {

packages/pg-query-stream/test/async-iterator.es6

@@ -59,7 +59,7 @@ describe('Async iterator', () => {
     const pool = new pg.Pool({ max: 1 })
     const client = await pool.connect()
     const rows = []
-    for await (const row of client.query(new QueryStream(queryText, []))) {
+    for await (const row of client.query(new QueryStream(queryText, [], { batchSize: 1 }))) {
       rows.push(row)
       break;
     }

packages/pg-query-stream/test/close.js

@@ -4,6 +4,10 @@ var concat = require('concat-stream')
 var QueryStream = require('../')
 var helper = require('./helper')
 
+if (process.version.startsWith('v8.')) {
+  return console.error('warning! node versions less than 10lts no longer supported & stream closing semantics may not behave properly');
+}
+
 helper('close', function (client) {
   it('emits close', function (done) {
     var stream = new QueryStream('SELECT * FROM generate_series(0, $1) num', [3], { batchSize: 2, highWaterMark: 2 })
@@ -23,27 +27,32 @@ helper('early close', function (client) {
       query.read()
     })
     query.once('readable', function () {
-      query.close()
+      query.destroy()
     })
     query.on('close', function () {
       assert(readCount < 10, 'should not have read more than 10 rows')
       done()
     })
   })
-})
 
-helper('close callback', function (client) {
-  it('notifies an optional callback when the conneciton is closed', function (done) {
-    var stream = new QueryStream('SELECT * FROM generate_series(0, $1) num', [10], { batchSize: 2, highWaterMark: 2 })
-    var query = client.query(stream)
-    query.once('readable', function () { // only reading once
-      query.read()
-    })
-    query.once('readable', function () {
-      query.close(function () {
-        // nothing to assert.  This test will time out if the callback does not work.
-        done()
-      })
+  it('can destroy stream while reading', function (done) {
+    var stream = new QueryStream('SELECT * FROM generate_series(0, 100), pg_sleep(1)')
+    client.query(stream)
+    stream.on('data', () => done(new Error('stream should not have returned rows')))
+    setTimeout(() => {
+      stream.destroy()
+      stream.on('close', done)
+    }, 100)
+  })
+
+  it('can destroy stream while reading an error', function (done) {
+    var stream = new QueryStream('SELECT * from pg_sleep(1), basdfasdf;')
+    client.query(stream)
+    stream.on('data', () => done(new Error('stream should not have returned rows')))
+    stream.once('error', () => {
+      stream.destroy()
+      // wait a bit to let any other errors shake through
+      setTimeout(done, 100)
     })
   })
 })