query events for prepared statements are handled by client

lib/client.js

@@ -86,8 +86,16 @@ p.connect = function() {
       self.activeQuery.handleDataRow(msg);
     });
 
+    con.on('portalSuspended', function(msg) {
+      self.activeQuery.getRows(con);
+    });
+
     con.on('commandComplete', function(msg) {
       self.activeQuery.handleCommandComplete(msg);
+      //need to sync after each command complete of a prepared statement
+      if(self.activeQuery.isPreparedStatement) {
+        con.sync();
+      }
     });
 
     con.on('readyForQuery', function() {
@@ -103,6 +111,10 @@ p.connect = function() {
       self.emit('error', error);
     } else {
       self.activeQuery.handleError(error);
+      //need to sync after error during a prepared statement
+      if(self.activeQuery.isPreparedStatement) {
+        con.sync();
+      }
       self.activeQuery = null;
     }
   });

lib/query.js

@@ -15,6 +15,7 @@ var Query = function(config) {
   this._fieldNames = [];
   this._fieldConverters = [];
   this._result = new Result();
+  this.isPreparedStatement = false;
   EventEmitter.call(this);
 };
 
@@ -125,9 +126,20 @@ p.hasBeenParsed = function(connection) {
   return this.name && connection.parsedStatements[this.name];
 };
 
+p.getRows = function(connection) {
+  connection.execute({
+    portal: this.name,
+    rows: this.rows
+  });
+  connection.flush();
+};
+
 p.prepare = function(connection) {
   var self = this;
-
+  //prepared statements need sync to be called after each command
+  //complete or when an error is encountered
+  this.isPreparedStatement = true;
+  //TODO refactor this poor encapsulation
   if(!this.hasBeenParsed(connection)) {
     connection.parse({
       text: self.text,
@@ -156,26 +168,7 @@ p.prepare = function(connection) {
     name: self.name || ""
   });
 
-  var getRows = function() {
-    connection.execute({
-      portal: self.name,
-      rows: self.rows
-    });
-    connection.flush();
-  };
-
-  getRows();
-
-  var onCommandComplete =  function() {
-    connection.removeListener('error', onCommandComplete);
-    connection.removeListener('commandComplete', onCommandComplete);
-    connection.removeListener('portalSuspended', getRows);
-    connection.sync();
-  };
-
-  connection.on('portalSuspended', getRows);
-  connection.on('commandComplete', onCommandComplete);
-  connection.on('error', onCommandComplete);
+  this.getRows(connection);
 };
 
 var dateParser = function(isoDate) {