Merge pull request #59 from brianc/bmc/fix-named-portal

Fix named portal not closing

.eslintrc

@@ -1,5 +1,8 @@
 {
   "extends": ["eslint:recommended"],
+  "parserOptions": {
+    "ecmaVersion": 2017
+  },
   "plugins": ["prettier"],
   "rules": {
     "prettier/prettier": "error",

.travis.yml

@@ -2,14 +2,14 @@ language: node_js
 dist: trusty
 sudo: false
 node_js:
-  - "8"
-  - "10"
-  - "12"
+  - '8'
+  - '10'
+  - '12'
 env:
   - PGUSER=postgres
 services:
   - postgresql
 addons:
-  postgresql: "9.6"
+  postgresql: '9.6'
 before_script:
   - psql -c 'create database travis;' -U postgres | true

index.js

@@ -75,6 +75,15 @@ Cursor.prototype._shiftQueue = function() {
   }
 }
 
+Cursor.prototype._closePortal = function() {
+  // because we opened a named portal to stream results
+  // we need to close the same named portal.  Leaving a named portal
+  // open can lock tables for modification if inside a transaction.
+  // see https://github.com/brianc/node-pg-cursor/issues/56
+  this.connection.close({ type: 'P', name: this._portal })
+  this.connection.sync()
+}
+
 Cursor.prototype.handleRowDescription = function(msg) {
   this._result.addFields(msg.fields)
   this.state = 'idle'
@@ -105,7 +114,7 @@ Cursor.prototype._sendRows = function() {
 
 Cursor.prototype.handleCommandComplete = function(msg) {
   this._result.addCommandComplete(msg)
-  this.connection.sync()
+  this._closePortal()
 }
 
 Cursor.prototype.handlePortalSuspended = function() {
@@ -114,8 +123,8 @@ Cursor.prototype.handlePortalSuspended = function() {
 
 Cursor.prototype.handleReadyForQuery = function() {
   this._sendRows()
-  this.emit('end', this._result)
   this.state = 'done'
+  this.emit('end', this._result)
 }
 
 Cursor.prototype.handleEmptyQuery = function() {
@@ -166,8 +175,7 @@ Cursor.prototype.close = function(cb) {
   if (this.state === 'done') {
     return setImmediate(cb)
   }
-  this.connection.close({ type: 'P' })
-  this.connection.sync()
+  this._closePortal()
   this.state = 'done'
   if (cb) {
     this.connection.once('closeComplete', function() {

test/transactions.js

@@ -0,0 +1,43 @@
+const assert = require('assert')
+const Cursor = require('../')
+const pg = require('pg')
+
+describe('transactions', () => {
+  it('can execute multiple statements in a transaction', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    await client.query('begin')
+    await client.query('CREATE TEMP TABLE foobar(id SERIAL PRIMARY KEY)')
+    const cursor = client.query(new Cursor('SELECT * FROM foobar'))
+    const rows = await new Promise((resolve, reject) => {
+      cursor.read(10, (err, rows) => (err ? reject(err) : resolve(rows)))
+    })
+    assert.equal(rows.length, 0)
+    await client.query('ALTER TABLE foobar ADD COLUMN name TEXT')
+    await client.end()
+  })
+
+  it('can execute multiple statements in a transaction if ending cursor early', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    await client.query('begin')
+    await client.query('CREATE TEMP TABLE foobar(id SERIAL PRIMARY KEY)')
+    const cursor = client.query(new Cursor('SELECT * FROM foobar'))
+    await new Promise(resolve => cursor.close(resolve))
+    await client.query('ALTER TABLE foobar ADD COLUMN name TEXT')
+    await client.end()
+  })
+
+  it.only('can execute multiple statements in a transaction if no data', async () => {
+    const client = new pg.Client()
+    await client.connect()
+    await client.query('begin')
+    // create a cursor that has no data response
+    const createText = 'CREATE TEMP TABLE foobar(id SERIAL PRIMARY KEY)'
+    const cursor = client.query(new Cursor(createText))
+    const err = await new Promise(resolve => cursor.read(100, resolve))
+    assert.ifError(err)
+    await client.query('ALTER TABLE foobar ADD COLUMN name TEXT')
+    await client.end()
+  })
+})