Add connection.dispatchQuery and connection.reset

binding.cc

@@ -22,6 +22,8 @@ class Connection : public EventEmitter {
     t->InstanceTemplate()->SetInternalFieldCount(1);
 
     NODE_SET_PROTOTYPE_METHOD(t, "connect", Connect);
+    NODE_SET_PROTOTYPE_METHOD(t, "reset", Reset);
+    NODE_SET_PROTOTYPE_METHOD(t, "dispatchQuery", DispatchQuery);
 
     t->PrototypeTemplate()->SetAccessor(READY_STATE_SYMBOL, ReadyStateGetter);
 
@@ -103,6 +105,11 @@ class Connection : public EventEmitter {
     Detach();
   }
 
+  char * ErrorMessage ( )
+  {
+    return PQerrorMessage(connection_);
+  }
+
  protected:
 
   static Handle<Value>
@@ -132,7 +139,49 @@ class Connection : public EventEmitter {
     bool r = connection->Connect(*conninfo);
 
     if (!r) {
-      return ThrowException(String::New("Error opening connection."));
+      return ThrowException(Exception::Error(
+            String::New(connection->ErrorMessage())));
+    }
+
+    return Undefined();
+  }
+
+  static Handle<Value>
+  Reset (const Arguments& args)
+  {
+    Connection *connection = ObjectWrap::Unwrap<Connection>(args.This());
+
+    HandleScope scope;
+
+    bool r = connection->Reset();
+
+    if (!r) {
+      return ThrowException(Exception::Error(
+            String::New(connection->ErrorMessage())));
+    }
+
+    return Undefined();
+  }
+
+  static Handle<Value>
+  DispatchQuery (const Arguments& args)
+  {
+    Connection *connection = ObjectWrap::Unwrap<Connection>(args.This());
+
+    HandleScope scope;
+
+    if (args.Length() == 0 || !args[0]->IsString()) {
+      return ThrowException(Exception::TypeError(
+            String::New("First argument must be a string")));
+    }
+
+    String::Utf8Value query(args[0]->ToString());
+
+    bool r = connection->Query(*query);
+
+    if (!r) {
+      return ThrowException(Exception::Error(
+            String::New(connection->ErrorMessage())));
     }
 
     return Undefined();
@@ -183,7 +232,7 @@ class Connection : public EventEmitter {
     return scope.Close(String::NewSymbol(s));
   }
 
-  Connection ( ) : EventEmitter () 
+  Connection () : EventEmitter () 
   {
     connection_ = NULL;
 
@@ -201,6 +250,7 @@ class Connection : public EventEmitter {
     assert(connection_ == NULL);
   }
 
+ private:
   void ConnectEvent ()
   {
     PostgresPollingStatusType status;
@@ -237,7 +287,6 @@ class Connection : public EventEmitter {
     Finish();
   }
 
- private:
   void EmitResult (PGresult *result)
   {
     switch (PQresultStatus(result)) {

postgres.js

@@ -1,18 +1,35 @@
-var ext = require("binding.node");
+var binding = require("binding.node");
 
-exports.createConnection = function (conninfo) {
-  var c = new ext.Connection;
-  c.connect(conninfo);
-  return c;
+var Connection = binding.Connection;
+
+// postgres cannot handle multiple queries at the same time.
+// thus we must queue them internally and dispatch them as
+// others come in. 
+Connection.prototype.maybeDispatchQuery = function () {
+  if (!this._queries) return;
+  // If not connected, do not dispatch. 
+  if (this.readyState != "OK") return;
+  if (this._queries.length > 0) this.dispatchQuery(this._queries[0]);
+};
+
+Connection.prototype.query = function (sql) {
+  if (!this._queries) this._queries = [];
+  this._queries.push(sql);
+  this.maybeDispatchQuery();
 };
 
-var c = exports.createConnection("host=/var/run/postgresql dbname=test");
+exports.createConnection = function (conninfo) {
+  var c = new Connection;
+
+  c.addListener("connect", function () {
+    c.maybeDispatchQuery();
+  });
+
+  c.addListener("result", function () {
+    c._queries.shift();
+  });
 
-c.addListener("connect", function () {
-  puts("connected");
-  puts(c.readyState);
-});
+  c.connect(conninfo);
 
-c.addListener("error", function () {
-  puts("error");
-});
+  return c;
+};

test.js

@@ -0,0 +1,14 @@
+var postgres = require("postgres.js");
+
+var c = postgres.createConnection("host=/var/run/postgresql dbname=test");
+
+c.addListener("connect", function () {
+  puts("connected");
+  puts(c.readyState);
+});
+
+c.addListener("error", function () {
+  puts("error");
+});
+
+c.query("select * from xxx;");