Merge pull request brianc#25 from sberan/cursor-result

brianc · web-flow · commit bbc2b416ed01 · 2017-05-15T23:15:59.000-05:00
Emit Query Events
diff --git a/.travis.yml b/.travis.yml
@@ -1,5 +1,13 @@
 language: node_js
+dist: trusty
+sudo: false
 node_js:
   - "6"
 env:
   - PGUSER=postgres
+services:
+  - postgresql
+addons:
+  postgresql: "9.6"
+before_script:
+  - psql -c 'create database travis;' -U postgres | true
diff --git a/index.js b/index.js
@@ -1,7 +1,11 @@
 var Result = require('./pg').Result
 var prepare = require('./pg').prepareValue
+var EventEmitter = require('events').EventEmitter;
+var util = require('util');
+
+function Cursor (text, values) {
+  EventEmitter.call(this);
 
-var Cursor = function(text, values) {
   this.text = text
   this.values = values ? values.map(prepare) : null
   this.connection = null
@@ -12,6 +16,8 @@ var Cursor = function(text, values) {
   this._rows = null
 }
 
+util.inherits(Cursor, EventEmitter)
+
 Cursor.prototype.submit = function(connection) {
   this.connection = connection
 
@@ -58,6 +64,7 @@ Cursor.prototype.handleRowDescription = function(msg) {
 
 Cursor.prototype.handleDataRow = function(msg) {
   var row = this._result.parseRow(msg.fields)
+  this.emit('row', row, this._result)
   this._rows.push(row)
 }
 
@@ -87,6 +94,7 @@ Cursor.prototype.handlePortalSuspended = function() {
 
 Cursor.prototype.handleReadyForQuery = function() {
   this._sendRows()
+  this.emit('end', this._result)
   this.state = 'done'
 }
 
@@ -107,6 +115,11 @@ Cursor.prototype.handleError = function(msg) {
   for(var i = 0; i < this._queue.length; i++) {
     this._queue.pop()[1](msg)
   }
+
+  if (this.eventNames().indexOf('error') >= 0) {
+    //only dispatch error events if we have a listener
+    this.emit('error', msg)
+  }
   //call sync to keep this connection from hanging
   this.connection.sync()
 }
diff --git a/test/index.js b/test/index.js
@@ -127,4 +127,33 @@ describe('cursor', function() {
       done()
     })
   })
+
+  it('emits row events', function(done) {
+    var cursor = this.pgCursor(text)
+    cursor.read(10)
+    cursor.on('row', (row, result) => result.addRow(row))
+    cursor.on('end', (result) => {
+      assert.equal(result.rows.length, 6)
+      done()
+    })
+  })
+
+  it('emits row events when cursor is closed manually', function(done) {
+    var cursor = this.pgCursor(text)
+    cursor.on('row', (row, result) => result.addRow(row))
+    cursor.on('end', (result) => {
+      assert.equal(result.rows.length, 3)
+      done()
+    })
+
+    cursor.read(3, () => cursor.close())
+  })
+
+  it('emits error events', function(done) {
+    var cursor = this.pgCursor('select asdfasdf')
+    cursor.on('error', function(err) {
+      assert(err)
+      done()
+    })
+  })
 })
