Catching and emitting errors from type parsers

index.js

@@ -75,9 +75,11 @@ Client.prototype.end = function(cb) {
   if(cb) setImmediate(cb);
 };
 
-Client.prototype._readError = function(message) {
+Client.prototype._readError = function(err) {
   this._stopReading();
-  var err = new Error(message || this.pq.errorMessage());
+  if(!(err instanceof Error)) {
+    err = new Error(err || this.pq.errorMessage());
+  }
   this.emit('error', err);
 };
 
@@ -109,7 +111,11 @@ Client.prototype._read = function() {
   var rows = []
   while(pq.getResult()) {
     if(pq.resultStatus() == 'PGRES_TUPLES_OK') {
-      this._parseResults(this.pq, rows);
+      try {
+        this._parseResults(this.pq, rows);
+      } catch (err) {
+        return this._readError(err);
+      }
     }
     if(pq.resultStatus() == 'PGRES_COPY_OUT')  break;
   }

test/query-async.js

@@ -2,6 +2,7 @@ var Client = require('../');
 var assert = require('assert');
 var async = require('async');
 var ok = require('okay');
+var PgTypes = require('pg-types');
 
 describe('async query', function() {
   before(function(done) {
@@ -14,6 +15,7 @@ describe('async query', function() {
 
   after(function(done) {
     this.client.end(done);
+    PgTypes.setTypeParser(23, 'text', null);
   });
 
   it('simple query works', function(done) {
@@ -86,4 +88,18 @@ describe('async query', function() {
       done();
     });
   });
+
+  it('returns an error if there was a error parsingRows', function(done) {
+    PgTypes.setTypeParser(23, 'text', function () {
+      throw new Error("Type Parser Error")
+    });
+    var runErrorQuery = function(n, done) {
+      this.client.query('SELECT 1', function(err) {
+        assert(err instanceof Error, 'Should return an error instance');
+        PgTypes.setTypeParser(23, 'text', null);
+        done();
+      });
+    }.bind(this);
+    async.timesSeries(3, runErrorQuery, done);
+  });
 });