conversion pg text to binary protocol started

Author: AlexanderS

lib/connection.js

@@ -148,7 +148,8 @@ p.bind = function(config) {
       buffer.addString(val);
     }
   }
-  buffer.addInt16(0); //no format codes, use text
+  buffer.addInt16(1); // format codes to use binary
+  buffer.addInt16(1);
   //0x42 = 'B'
   this.send(0x42, buffer.flush());
 };
@@ -365,7 +366,7 @@ p.parseD = function(msg) {
   var fields = [];
   for(var i = 0; i < fieldCount; i++) {
     var length = this.parseInt32();
-    fields[i] = (length === -1 ? null : this.readString(length))
+    fields[i] = (length === -1 ? null : this.readBytes(length))
   };
   msg.fieldCount = fieldCount;
   msg.fields = fields;
@@ -434,6 +435,10 @@ p.readString = function(length) {
   return this.buffer.toString(this.encoding, this.offset, (this.offset += length));
 };
 
+p.readBytes = function(length) {
+  return this.buffer.slice(this.offset, this.offset += length);
+};
+
 p.parseCString = function() {
   var start = this.offset;
   while(this.buffer[this.offset++]) { };

lib/query.js

@@ -37,6 +37,7 @@ p.submit = function(connection) {
   var names = [];
   var rows = [];
   var handleRowDescription = function(msg) {
+      console.log(JSON.stringify(msg));
     for(var i = 0; i < msg.fields.length; i++) {
       converters[i] = dataTypeParsers[msg.fields[i].dataTypeID] || noParse;
       names[i] = msg.fields[i].name;
@@ -47,6 +48,7 @@ p.submit = function(connection) {
     for(var i = 0; i < msg.fields.length; i++) {
       var rawValue = msg.fields[i];
       result[names[i]] = rawValue === null ? null : converters[i](rawValue);
+      console.log(names[i] + ": " + result[names[i]]);
     }
     self.emit('row', result);
 
@@ -206,22 +208,135 @@ var dateParser = function(isoDate) {
   return date;
 };
 
+function shl(a,b) {
+    // Copyright (c) 1996 Henri Torgemane. All Rights Reserved.
+    // fix for crappy <<
+    for (var i=0;i<b;i++) {
+        a=a%0x80000000;
+        if (a&0x40000000==0x40000000)
+        {
+            a-=0x40000000;
+            a*=2;
+            a+=0x80000000;
+        } else
+            a*=2;
+    };
+
+    return a;
+}
+
+var parseFloat = function(data, precisionBits, exponentBits) {
+    var bias = Math.pow(2, exponentBits - 1) - 1;
+    var sign = parseBits(data, 1);
+    var exponent = parseBits(data, exponentBits, 1);
+
+    if (exponent == 0)
+        return 0;
+
+    // parse mantissa
+    var precisionBitsCounter = 1;
+    var parsePrecisionBits = function(lastValue, newValue, bits) {
+        if (lastValue == 0) {
+            lastValue = 1;
+        }
+
+        for (var i = 1; i <= bits; i++) {
+            precisionBitsCounter /= 2;
+            if ((newValue & (0x1 << (bits - i))) > 0) {
+                lastValue += precisionBitsCounter;
+            }
+        }
+
+        return lastValue;
+    };
+
+    var mantissa = parseBits(data, precisionBits, exponentBits + 1, parsePrecisionBits);
+
+    // special cases
+    if (exponent == (Math.pow(2, exponentBits + 1) - 1)) {
+        if (mantissa == 0) {
+            return (sign == 0) ? Infinity : -Infinity;
+        }
+
+        return NaN;
+    }
+
+    // normale number
+    return ((sign == 0) ? 1 : -1) * Math.pow(2, exponent - bias) * mantissa;
+};
+
+var parseBits = function(data, bits, offset, callback) {
+    offset = offset || 0;
+    callback = callback || function(lastValue, newValue, bits) { return (lastValue * Math.pow(2, bits)) + newValue; };
+    var offsetBytes = offset >> 3;
+
+    // read first (maybe partial) byte
+    var mask = 0xff;
+    var firstBits = 8 - (offset % 8);
+    if (bits < firstBits) {
+        mask = (0xff << (8 - bits)) & 0xff;
+        firstBits = bits;
+    }
+
+    if (offset) {
+        mask = mask >> (offset % 8);
+    }
+    var result = callback(0, data[offsetBytes] & mask, firstBits);
+
+    // read bytes
+    var bytes = (bits + offset) >> 3;
+    for (var i = offsetBytes + 1; i < bytes; i++) {
+        result = callback(result, data[i], 8);
+    }
+
+    // bits to read, that are not a complete byte
+    var lastBits = (bits + offset) % 8;
+    if (lastBits > 0) {
+        result = callback(result, data[bytes] >> (8 - lastBits), lastBits);
+    }
+
+    return result;
+}
+
+var parseBinaryInt64 = function(value) {
+    return parseBits(value, 64);
+}
+
+var parseBinaryInt32 = function(value) {
+    return parseBits(value, 32);
+}
+
+var parseBinaryInt16 = function(value) {
+    return parseBits(value, 16);
+}
+
+var parseBinaryFloat32 = function(value) {
+    return parseFloat(value, 23, 8);
+}
+
+var parseBinaryFloat64 = function(value) {
+    return parseFloat(value, 52, 11);
+}
+
 // To help we test dateParser
 Query.dateParser = dateParser;
 
 var dataTypeParsers = {
-  20: parseInt,
-  21: parseInt,
-  23: parseInt,
-  26: parseInt,
-  1700: parseFloat,
-  700: parseFloat,
-  701: parseFloat,
   16: function(dbVal) { //boolean
-    return dbVal === 't';
+      console.log(JSON.stringify(dbVal));
+      return value[0] == 1;
   },
-  1114: dateParser,
+
+  20: parseBinaryInt64,
+  21: parseBinaryInt16,
+  23: parseBinaryInt32,
+  26: parseBinaryInt64,
+  700: parseBinaryFloat32,
+  701: parseBinaryFloat64,
+//  1009: arrayParser,
+    1114: parseBinaryInt64, // TOFIX: dateParser,
   1184: dateParser
+  // 1700: parseFloat,
 };
 
 

lib/result.js

@@ -0,0 +1,32 @@
+//result object returned from query
+//in the 'end' event and also
+//passed as second argument to provided callback
+var Result = function() {
+  this.rows = [];
+};
+
+var p = Result.prototype;
+
+
+var matchRegexp = /([A-Za-z]+) (\d+ )?(\d+)?/
+
+//adds a command complete message
+p.addCommandComplete = function(msg) {
+  var match = matchRegexp.exec(msg.text);
+  if(match) {
+    this.command = match[1];
+    //match 3 will only be existing on insert commands
+    if(match[3]) {
+      this.rowCount = parseInt(match[3]);
+      this.oid = parseInt(match[2]);
+    } else {
+      this.rowCount = parseInt(match[2]);
+    }
+  }
+};
+
+p.addRow = function(row) {
+  this.rows.push(row);
+};
+
+module.exports = Result;

lib/writer.js

@@ -1,25 +1,18 @@
 var Writer = function(size) {
   this.size = size || 1024;
-  this.buffer = new Buffer(this.size);
-  this.offset = 0;
+  this.buffer = new Buffer(this.size + 5);
+  this.offset = 5;
 };
 
 var p = Writer.prototype;
 
-p._remaining = function() {
-  return this.buffer.length - this.offset;
-}
-
-p._resize = function() {
-  var oldBuffer = this.buffer;
-  this.buffer = Buffer(oldBuffer.length + this.size);
-  oldBuffer.copy(this.buffer);
-}
-
 //resizes internal buffer if not enough size left
 p._ensure = function(size) {
-  if(this._remaining() < size) {
-    this._resize()
+  var remaining = this.buffer.length - this.offset;
+  if(remaining < size) {
+    var oldBuffer = this.buffer;
+    this.buffer = Buffer(oldBuffer.length + size);
+    oldBuffer.copy(this.buffer);
   }
 }
 
@@ -46,7 +39,7 @@ p.addCString = function(string) {
   this.buffer.write(string, this.offset);
   this.offset += len;
   this.buffer[this.offset] = 0; //add null terminator
-  return this;  
+  return this;
 }
 
 p.addChar = function(char) {
@@ -56,10 +49,6 @@ p.addChar = function(char) {
   return this;
 }
 
-p.join = function() {
-  return this.buffer.slice(0, this.offset);
-}
-
 p.addString = function(string) {
   var string = string || "";
   var len = Buffer.byteLength(string);
@@ -70,7 +59,7 @@ p.addString = function(string) {
 }
 
 p.getByteLength = function() {
-  return this.offset;
+  return this.offset - 5;
 }
 
 p.add = function(otherBuffer) {
@@ -81,11 +70,24 @@ p.add = function(otherBuffer) {
 }
 
 p.clear = function() {
-  this.offset=0;
+  this.offset=5;
+}
+
+p.join = function(code) {
+  if(code) {
+    var end = this.offset;
+    this.offset = 0;
+    this.buffer[this.offset++] = code;
+    //write the length which is length of entire packet not including
+    //message type code byte
+    this.addInt32(end - 1);
+    this.offset = end;
+  }
+  return this.buffer.slice(code ? 0 : 5, this.offset);
 }
 
-p.flush = function() {
-  var result = this.join();
+p.flush = function(code) {
+  var result = this.join(code);
   this.clear();
   return result;
 }

package.json

@@ -1,5 +1,5 @@
 { "name": "pg",
-  "version": "0.2.3",
+  "version": "0.2.6",
   "description": "Pure JavaScript PostgreSQL client",
   "homepage": "http://github.com/brianc/node-postgres",
   "repository" : {
@@ -9,6 +9,6 @@
   "author" : "Brian Carlson <brian.m.carlson@gmail.com>",
   "main" : "./lib/index",
   "directories" : { "lib" : "./lib" },
-  "scripts" : { "test" : "node ./test/run.js" },
+  "scripts" : { "test" : "make test" },
   "engines" : { "node": ">= 0.2.2" }
 }

script/list-db-types.js

@@ -1,24 +1,6 @@
-var net = require('net')
-var Connection = require(__dirname+'/../lib/connection');
-
-var con = new Connection({stream: new net.Stream()});
-con.connect('5432', 'localhost');
-
-con.on('connect', function() {
-  con.startup({
-    user: 'brian',
-    database: 'postgres'
-  });
-});
-
-con.on('dataRow', function(msg) {
-  console.log(msg.fields);
-});
-
-con.on('readyForQuery', function() {
-  con.query('select oid, typname from pg_type where typtype = \'b\' order by typname');
-});
-
-con.on('commandComplete', function() {
-  con.end();
-});
+var helper = require(__dirname + "/../test/integration/test-helper");
+var pg = helper.pg;
+pg.connect(helper.connectionString(), assert.success(function(client) {
+  var query = client.query('select oid, typname from pg_type where typtype = \'b\' order by oid');
+  query.on('row', console.log);
+}))

test/integration/client/api-tests.js

@@ -6,7 +6,7 @@ var log = function() {
   //console.log.apply(console, arguments);
 }
 
-var sink = new helper.Sink(4, 10000, function() {
+var sink = new helper.Sink(5, 10000, function() {
   log("ending connection pool: %s", connectionString);
   pg.end(connectionString);
 });
@@ -92,3 +92,19 @@ test("query errors are handled and do not bubble if callback is provded", functi
    }))
   }))
 })
+
+test('callback is fired once and only once', function() {
+  pg.connect(connectionString, assert.calls(function(err, client) {
+    assert.isNull(err);
+    client.query("CREATE TEMP TABLE boom(name varchar(10))");
+    var callCount = 0;
+    client.query([
+      "INSERT INTO boom(name) VALUES('hai')",
+      "INSERT INTO boom(name) VALUES('boom')",
+      "INSERT INTO boom(name) VALUES('zoom')",
+    ].join(";"), function(err, callback) {
+      assert.equal(callCount++, 0, "Call count should be 0.  More means this callback fired more than once.");
+      sink.add();
+    })
+  }))
+})