First tests passing for new native bindings

Author: brianc

lib/native/index.js

@@ -1,3 +1,100 @@
+var Native = require('pg-native');
+var EventEmitter = require('events').EventEmitter;
+var util = require('util');
+
+var NativeQuery = require('./query');
+
+var Client = module.exports = function() {
+  EventEmitter.call(this);
+  this.native = new Native();
+  this._queryQueue = [];
+};
+
+util.inherits(Client, EventEmitter);
+
+//connect to the backend
+//pass an optional callback to be called once connected
+//or with an error if there was a connection error
+//if no callback is passed and there is a connection error
+//the client will emit an error event.
+Client.prototype.connect = function(cb) {
+  var self = this;
+  this.native.connect(function(err) {
+    //error handling
+    if(err) {
+      if(cb) return cb(err);
+      return self.emit('error')
+    }
+
+    //set internal states to connected
+    self._connected = true;
+    self.emit('connect');
+    self._pulseQueryQueue(true);
+
+    //possibly call the optional callback
+    if(cb) cb();
+  });
+};
+
+Client.prototype.query = function(config, values, callback) {
+  var query = new NativeQuery(this.native);
+
+  //support query('text', ...) style calls
+  if(typeof config == 'string') {
+    query.text = config;
+  }
+
+  //support passing everything in via a config object
+  if(typeof config == 'object') {
+    query.text = config.text;
+    query.values = config.values;
+    query.name = config.name;
+    query.callback = config.callback;
+  }
+
+  //support query({...}, function() {}) style calls
+  //& support query(..., ['values'], ...) style calls
+  if(typeof values == 'function') {
+    query.callback = values;
+  }
+  else if(util.isArray(values)) {
+    query.values = values;
+  }
+  if(typeof callback == 'function') {
+    query.callback = callback;
+  }
+
+  this._queryQueue.push(query);
+  this._pulseQueryQueue();
+  return query;
+};
+
+Client.prototype.end = function(cb) {
+  this.native.end(cb);
+};
+
+Client.prototype._pulseQueryQueue = function(initialConnection) {
+  if(!this._connected) {
+    return;
+  }
+  if(this._activeQuery) {
+    if(this._activeQuery.state != 'error' && this._activeQuery.state != 'done') {
+      return;
+    }
+  }
+  var query = this._queryQueue.shift();
+  if(!query) {
+    if(!initialConnection) {
+      this.emit('drain');
+    }
+    return;
+  }
+  this._activeQuery = query;
+  query.submit();
+};
+
+
+return;
 //require the c++ bindings & export to javascript
 var EventEmitter = require('events').EventEmitter;
 
@@ -108,38 +205,6 @@ Connection.prototype.cancel = function(client, query) {
   }
 };
 
-Connection.prototype._pulseQueryQueue = function(initialConnection) {
-  if(!this._connected) {
-    return;
-  }
-  if(this._activeQuery) {
-    return;
-  }
-  var query = this._queryQueue.shift();
-  if(!query) {
-    if(!initialConnection) {
-      this.emit('drain');
-    }
-    return;
-  }
-  this._activeQuery = query;
-  if(query.name) {
-    if(this._namedQueries[query.name]) {
-      this._sendQueryPrepared(query.name, query.values||[], query.singleRowMode);
-    } else {
-      this._namedQuery = true;
-      this._namedQueries[query.name] = true;
-      this._sendPrepare(query.name, query.text, (query.values||[]).length, query.singleRowMode);
-    }
-  } else if(query.values) {
-    //call native function
-    this._sendQueryWithParams(query.text, query.values, query.singleRowMode);
-  } else {
-    //call native function
-    this._sendQuery(query.text, query.singleRowMode);
-  }
-};
-
 Connection.prototype.sendCopyFail = function(msg) {
   this.endCopyFrom(msg);
 };

lib/native/query.js

@@ -1,6 +1,48 @@
 var EventEmitter = require('events').EventEmitter;
 var util = require('util');
 
+var NativeQuery = module.exports = function(native) {
+  EventEmitter.call(this);
+  this.native = native;
+  this.text = null;
+  this.values = null;
+  this.name = null;
+  this.callback = null;
+  this.state = 'new';
+};
+
+util.inherits(NativeQuery, EventEmitter);
+
+NativeQuery.prototype.submit = function() {
+  this.state = 'running';
+  var self = this;
+
+  var after = function(err, rows) {
+
+    //handle possible query error
+    if(err) {
+      self.state = 'error';
+      if(self.callback) return self.callback(err);
+      return self.emit('error', err);
+    }
+
+    //handle successful result
+    self.state = 'done';
+    self.emit('done');
+    if(self.callback) {
+      self.callback(null, {rows: rows})
+    }
+  }
+
+  if(this.values) {
+    this.native.query(this.text, this.values, after);
+  } else {
+    this.native.query(this.text, after);
+  }
+};
+
+return;
+
 var utils = require(__dirname + '/../utils');
 var Result = require(__dirname + '/../result');