Emit Query Events

sberan · sberan · commit acae15de53a2 · 2017-05-09T10:13:57.000-05:00
This change adds events to the `Cursor` object as per the [Query API](https://github.com/brianc/node-postgres/wiki/Query).
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,7 @@ Cursor.prototype.handleError = function(msg) {
   for(var i = 0; i < this._queue.length; i++) {
     this._queue.pop()[1](msg)
   }
+  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()
+    })
+  })
 })
