COPY TO/FROM native/libpq done. Looks like it works, but need to test

lib/client.js

@@ -6,7 +6,8 @@ var Query = require(__dirname + '/query');
 var utils = require(__dirname + '/utils');
 var defaults = require(__dirname + '/defaults');
 var Connection = require(__dirname + '/connection');
-
+var CopyFromStream = require(__dirname + '/copystream').CopyFromStream;
+var CopyToStream = require(__dirname + '/copystream').CopyToStream;
 var Client = function(config) {
   EventEmitter.call(this);
   if(typeof config === 'string') {
@@ -104,7 +105,12 @@ p.connect = function(callback) {
         con.sync();
       }
     });
-
+    con.on('copyInResponse', function(msg) {
+      self.activeQuery.streamData(self.connection);
+    });
+    con.on('copyData', function (msg) {
+      self.activeQuery.handleCopyFromChunk(msg.chunk);
+    });
     if (!callback) {
       self.emit('connect');
     } else {
@@ -184,7 +190,30 @@ p._pulseQueryQueue = function() {
     }
   }
 };
+p._copy = function (text, stream) {
+ var config = {},
+    query;
+  config.text = text;
+  config.stream = stream;
+  config.callback = function (error) {
+    if (error) {
+      config.stream.error(error);
+    } else {
+      config.stream.close();
+    }
+  }
+  query = new Query(config);
+  this.queryQueue.push(query);
+  this._pulseQueryQueue();
+  return config.stream;
 
+};
+p.copyFrom = function (text) {
+  return this._copy(text, new CopyFromStream());
+}
+p.copyTo = function (text) {
+  return this._copy(text, new CopyToStream());
+}  
 p.query = function(config, values, callback) {
   //can take in strings, config object or query object
   var query = (config instanceof Query) ? config : new Query(config, values, callback);

lib/connection.js

@@ -261,7 +261,12 @@ p.describe = function(msg, more) {
   this.writer.addCString(msg.type + (msg.name || ''));
   this._send(0x44, more);
 };
-
+p.sendCopyFromChunk = function (chunk) {
+  this.stream.write(this.writer.add(chunk).flush(0x64));
+}
+p.endCopyFrom = function () {
+  this.stream.write(this.writer.add(emptyBuffer).flush(0x63));
+}
 //parsing methods
 p.setBuffer = function(buffer) {
   if(this.lastBuffer) {    //we have unfinished biznaz
@@ -311,7 +316,6 @@ p.parseMessage =  function() {
   var msg = {
     length: length
   };
-
   switch(id)
   {
 
@@ -375,6 +379,21 @@ p.parseMessage =  function() {
     msg.name = 'portalSuspended';
     return msg;
 
+  case 0x47: //G
+    msg.name = 'copyInResponse';
+    return this.parseGH(msg);
+
+  case 0x48: //H
+    msg.name = 'copyOutResponse';
+    return this.parseGH(msg); 
+  case 0x63: //c
+    msg.name = 'copyDone';
+    return  msg;
+
+  case 0x64: //d
+    msg.name = 'copyData';
+    return this.parsed(msg);
+
   default:
     throw new Error("Unrecognized message code " + id);
   }
@@ -505,7 +524,20 @@ p.parseA = function(msg) {
   msg.payload = this.parseCString();
   return msg;
 };
-
+p.parseGH = function (msg) {
+  msg.binary = Boolean(this.parseInt8());
+  var columnCount = this.parseInt16();
+  msg.columnTypes = [];
+  for(var i = 0; i<columnCount; i++) {
+    msg.columnTypes[i] = this.parseInt16();
+  }
+  return msg;
+};
+p.parseInt8 = function () {
+  var value = Number(this.buffer[this.offset]); 
+  this.offset++;
+  return value;
+}
 p.readChar = function() {
   return Buffer([this.buffer[this.offset++]]).toString(this.encoding);
 };
@@ -544,5 +576,10 @@ p.parseCString = function() {
   while(this.buffer[this.offset++]) { };
   return this.buffer.toString(this.encoding, start, this.offset - 1);
 };
+p.parsed = function (msg) {
+  //exclude length field
+  msg.chunk = this.readBytes(msg.length - 4);
+  return msg;  
+}
 //end parsing methods
 module.exports = Connection;

lib/copystream.js

@@ -0,0 +1,166 @@
+var Stream = require('stream').Stream;
+var util = require('util');
+var CopyFromStream = function () {
+  Stream.apply(this, arguments);
+  this._buffer = new Buffer(0);
+  this._connection = false;
+  this._finished  = false;
+  this._error = false;
+  this.__defineGetter__("writable", this._writable.bind(this));
+};
+util.inherits(CopyFromStream, Stream);
+CopyFromStream.prototype._writable = function () {
+  return !this._finished && !this._error;
+}
+CopyFromStream.prototype.startStreamingToConnection = function (connection) {
+  this._connection = connection;
+  this._handleChunk();
+  this._endIfConnectionReady();
+};
+CopyFromStream.prototype._handleChunk = function (string, encoding) {
+  var dataChunk,
+    tmpBuffer;
+  if (string !== undefined) {
+    if (string instanceof Buffer) {
+      dataChunk = string;
+    } else {
+      dataChunk = new Buffer(string, encoding);
+    }
+    if (this._buffer.length) {
+      //Buffer.concat is better, but it's missing 
+      //in node v0.6.x
+      tmpBuffer = new Buffer(this._buffer.length + dataChunk.length); 
+      tmpBuffer.copy(this._buffer); 
+      tmpBuffer.copy(dataChunk, this._buffer.length);
+      this._buffer = tmpBuffer;
+    } else {
+      this._buffer = dataChunk;
+    }
+  }
+  return this._sendIfConnectionReady();
+};
+CopyFromStream.prototype._sendIfConnectionReady = function () {
+  var dataSent = false;
+  if (this._connection && this._buffer.length) {
+    dataSent = this._connection.sendCopyFromChunk(this._buffer);
+    this._buffer = new Buffer(0);
+  }
+  return dataSent;
+};
+CopyFromStream.prototype._endIfConnectionReady = function () {
+  if (this._connection && this._finished) {
+    //TODO change function name
+    this._connection.endCopyFrom();
+  }
+}
+CopyFromStream.prototype.write = function (string, encoding) {
+  if (!this._writable) {
+    //TODO possibly throw exception?
+    return false;
+  }
+  return this._handleChunk.apply(this, arguments);
+};
+CopyFromStream.prototype.end = function (string, encondig) {
+  if(!this._writable) {
+    //TODO possibly throw exception?
+    return false;
+  }
+  this._finished = true;
+  if (string !== undefined) {
+    this._handleChunk.apply(this, arguments);
+  };  
+  this._endIfConnectionReady();
+};
+CopyFromStream.prototype.error = function (error) {
+  this._error = true;
+  this.emit('error', error);
+};
+CopyFromStream.prototype.close = function () {
+  this.emit("close");
+};
+var CopyToStream = function () {
+  Stream.apply(this, arguments);
+  this._error = false;
+  this._finished = false;
+  this._paused = false;
+  this.buffer = new Buffer(0);
+  this._encoding = undefined;
+  this.__defineGetter__('readable', this._readable.bind(this));
+};
+util.inherits(CopyToStream, Stream);
+CopyToStream.prototype._outputDataChunk = function () {
+  if (this._paused) {
+    return;
+  }
+  if (this.buffer.length) {
+    if (this._encoding) {
+      this.emit('data', this.buffer.toString(encoding));
+    } else {
+      this.emit('data', this.buffer);
+    }
+    this.buffer = new Buffer(0);
+  } 
+};
+CopyToStream.prototype._readable = function () {
+  return !this._finished && !this._error;
+}
+CopyToStream.prototype.error = function (error) {
+  if (!this.readable) {
+    return false;
+  }
+  this._error = error;
+  if (!this._paused) {
+    this.emit('error', error);
+  }
+};
+CopyToStream.prototype.close = function () {
+  if (!this.readable) {
+    return false;
+  }
+  this._finished = true;
+  if (!this._paused) {
+    this.emit("end");
+  }
+};
+CopyToStream.prototype.handleChunk = function (chunk) {
+  var tmpBuffer;
+  if (!this.readable) {
+    return;
+  }
+  if (!this.buffer.length) {
+    this.buffer = chunk;
+  } else {
+    tmpBuffer = new Buffer(this.buffer.length + chunk.length);
+    this.buffer.copy(tmpBuffer);
+    chunk.copy(tmpBuffer, this.buffer.length);
+    this.buffer = tmpBuffer;
+  }
+  this._outputDataChunk();
+};
+CopyToStream.prototype.pause = function () {
+  if (!this.readable) {
+    return false;
+  }
+  this._paused = true;
+};
+CopyToStream.prototype.resume = function () {
+  if (!this._paused) {
+    return false;
+  }
+  this._paused = false;
+  this._outputDataChunk();
+  if (this._error) {
+    return this.emit('error', this._error);
+  } 
+  if (this._finished) {
+    return this.emit('end');
+  }
+};
+CopyToStream.prototype.setEncoding = function (encoding) {
+  this._encoding = encoding;
+};
+
+module.exports = {
+  CopyFromStream: CopyFromStream,
+  CopyToStream: CopyToStream
+};

lib/native/index.js

@@ -1,6 +1,8 @@
 //require the c++ bindings & export to javascript
 var EventEmitter = require('events').EventEmitter;
 var utils = require(__dirname + "/../utils");
+var CopyFromStream = require(__dirname + '/../copystream').CopyFromStream;
+var CopyToStream = require(__dirname + '/../copystream').CopyToStream;
 
 var binding;
 
@@ -48,7 +50,31 @@ p.connect = function(cb) {
     nativeConnect.call(self, conString);
   })
 }
-
+p._copy = function (text, stream) {
+  var q = new NativeQuery(text, function (error) {
+    if (error) {
+      q.stream.error(error);
+    } else {
+      q.stream.close();
+    }
+  }); 
+  q.stream = stream;
+  this._queryQueue.push(q);
+  this._pulseQueryQueue();
+  return q.stream;
+}
+p.copyFrom = function (text) {
+  return this._copy(text, new CopyFromStream());
+};
+p.copyTo = function (text) {
+  return this._copy(text, new CopyToStream());
+};
+p.sendCopyFromChunk = function (chunk) {
+  this._sendCopyFromChunk(chunk); 
+};
+p.endCopyFrom = function () {
+  this._endCopyFrom();
+};
 p.query = function(config, values, callback) {
   var query = (config instanceof NativeQuery) ? config : new NativeQuery(config, values, callback);
   this._queryQueue.push(query);
@@ -167,7 +193,12 @@ var clientBuilder = function(config) {
       connection._pulseQueryQueue();
     }
   });
-
+  connection.on('_copyInResponse', function () {
+    connection._activeQuery.streamData(connection);
+  });
+  connection.on('_copyData', function (chunk) {
+    connection._activeQuery.handleCopyFromChunk(chunk);
+  });
   return connection;
 };
 

lib/native/query.js

@@ -67,5 +67,10 @@ p.handleReadyForQuery = function(meta) {
   }
   this.emit('end', this._result);
 };
-
+p.streamData = function (connection) {
+  this.stream.startStreamingToConnection(connection);
+};
+p.handleCopyFromChunk = function (chunk) {
+  this.stream.handleChunk(chunk);    
+}
 module.exports = NativeQuery;

lib/query.js

@@ -17,6 +17,7 @@ var Query = function(config, values, callback) {
   this.types = config.types;
   this.name = config.name;
   this.binary = config.binary;
+  this.stream  = config.stream;
   //use unique portal name each time
   this.portal = config.portal || ""
   this.callback = config.callback;
@@ -168,5 +169,10 @@ p.prepare = function(connection) {
 
   this.getRows(connection);
 };
-
+p.streamData = function (connection) {
+  this.stream.startStreamingToConnection(connection);
+};
+p.handleCopyFromChunk = function (chunk) {
+  this.stream.handleChunk(chunk);    
+}
 module.exports = Query;

src/binding.cc

@@ -1,5 +1,6 @@
 #include <libpq-fe.h>
 #include <node.h>
+#include <node_buffer.h>
 #include <string.h>
 #include <assert.h>
 #include <stdlib.h>
@@ -65,7 +66,6 @@ public:
     payload_symbol = NODE_PSYMBOL("payload");
     command_symbol = NODE_PSYMBOL("command");
 
-
     NODE_SET_PROTOTYPE_METHOD(t, "connect", Connect);
     NODE_SET_PROTOTYPE_METHOD(t, "_sendQuery", SendQuery);
     NODE_SET_PROTOTYPE_METHOD(t, "_sendQueryWithParams", SendQueryWithParams);
@@ -73,7 +73,8 @@ public:
     NODE_SET_PROTOTYPE_METHOD(t, "_sendQueryPrepared", SendQueryPrepared);
     NODE_SET_PROTOTYPE_METHOD(t, "cancel", Cancel);
     NODE_SET_PROTOTYPE_METHOD(t, "end", End);
-
+    NODE_SET_PROTOTYPE_METHOD(t, "_sendCopyFromChunk", SendCopyFromChunk);
+    NODE_SET_PROTOTYPE_METHOD(t, "_endCopyFrom", EndCopyFrom);
     target->Set(String::NewSymbol("Connection"), t->GetFunction());
     TRACE("created class");
   }
@@ -246,12 +247,13 @@ public:
   PGconn *connection_;
   bool connecting_;
   bool ioInitialized_;
+  bool copyOutMode_;
   Connection () : ObjectWrap ()
   {
     connection_ = NULL;
     connecting_ = false;
     ioInitialized_ = false;
-
+    copyOutMode_ = false;
     TRACE("Initializing ev watchers");
     read_watcher_.data = this;
     write_watcher_.data = this;
@@ -261,6 +263,26 @@ public:
   {
   }
 
+  static Handle<Value>
+  SendCopyFromChunk(const Arguments& args) {
+    HandleScope scope;
+    Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
+    //TODO handle errors in some way
+    if (args.Length() < 1 && !Buffer::HasInstance(args[0])) {
+      THROW("SendCopyFromChunk requires 1 Buffer argument");  
+    }
+    self->SendCopyFromChunk(args[0]->ToObject());
+    return Undefined();
+  }
+  static Handle<Value>
+  EndCopyFrom(const Arguments& args) {
+    HandleScope scope;
+    Connection *self = ObjectWrap::Unwrap<Connection>(args.This());
+    //TODO handle errors in some way
+    self->EndCopyFrom();
+    return Undefined();
+  }
+
 protected:
   //v8 entry point to constructor
   static Handle<Value>
@@ -408,14 +430,27 @@ protected:
       //declare handlescope as this method is entered via a libuv callback
       //and not part of the public v8 interface
       HandleScope scope;
-
+      if (this->copyOutMode_) {
+        this->HandleCopyOut();
+      }
       if (PQisBusy(connection_) == 0) {
         PGresult *result;
         bool didHandleResult = false;
         while ((result = PQgetResult(connection_))) {
-          HandleResult(result);
-          didHandleResult = true;
-          PQclear(result);
+          if (PGRES_COPY_IN == PQresultStatus(result)) {
+            didHandleResult = false;
+            Emit("_copyInResponse");
+            PQclear(result);
+            break;
+          } else if (PGRES_COPY_OUT == PQresultStatus(result)) {
+            PQclear(result);
+            this->copyOutMode_ = true;
+            didHandleResult = this->HandleCopyOut();
+          } else {
+            HandleResult(result);
+            didHandleResult = true;
+            PQclear(result);
+          }
         }
         //might have fired from notification
         if(didHandleResult) {
@@ -442,7 +477,37 @@ protected:
       }
     }
   }
-
+  bool HandleCopyOut () {
+    char * buffer = NULL;
+    int copied = PQgetCopyData(connection_, &buffer, 1);
+    if (copied > 0) {
+      Buffer * chunk = Buffer::New(buffer, copied);
+      Handle<Value> node_chunk = chunk->handle_; 
+      Emit("_copyData", &node_chunk);
+      PQfreemem(buffer);
+      //result was not handled copmpletely
+      return false;
+    } else if (copied == 0) {
+      //wait for next read ready
+      //result was not handled copmpletely
+      return false;
+    } else if (copied == -1) {
+      PGresult *result;
+      //result is handled completely
+      this->copyOutMode_ = false;
+      if (PQisBusy(connection_) == 0 && (result = PQgetResult(connection_))) {
+        HandleResult(result);
+        PQclear(result);
+        return true;
+      } else {
+        return false;
+      }
+    } else if (copied == -2) {
+      //TODO error handling
+      //result is handled with error
+      return true;
+    }
+  }
   void HandleResult(PGresult* result)
   {
     ExecStatusType status = PQresultStatus(result);
@@ -703,6 +768,13 @@ private:
     strcpy(cString, *utf8String);
     return cString;
   }
+  void SendCopyFromChunk(Handle<Object> chunk) {
+    PQputCopyData(connection_, Buffer::Data(chunk), Buffer::Length(chunk));
+  }
+  void EndCopyFrom() {
+    PQputCopyEnd(connection_, NULL);
+  }
+
 };