Update postgres.js to use the new bits library.

Author: creationix

postgres.js

@@ -1,30 +1,13 @@
 /*jslint bitwise: true, eqeqeq: true, immed: true, newcap: true, nomen: true, onevar: true, plusplus: true, regexp: true, undef: true, white: true, indent: 2 */
 /*globals include md5 node exports */
 
-process.mixin(require('./postgres-js/bits'));
 process.mixin(require('./postgres-js/md5'));
+var bits = require('./postgres-js/bits');
 var tcp = require("tcp");
 var sys = require("sys");
 
 exports.DEBUG = 0;
 
-String.prototype.add_header = function (code) {
-  if (code === undefined) {
-    code = "";
-  }
-  return code.add_int32(this.length + 4) + this;
-};
-
-Object.prototype.map_pairs = function (func) {
-  var result = [];
-  for (var k in this) {
-    if (this.hasOwnProperty(k)) {
-      result.push(func.call(this, k, this[k]));
-    }
-  }
-  return result;
-}
-
 // http://www.postgresql.org/docs/8.3/static/protocol-message-formats.html
 var formatter = {
   CopyData: function () {
@@ -34,62 +17,65 @@ var formatter = {
     // TODO: implement
   },
   Describe: function (name, type) {
-    var stream = [type.charCodeAt(0)].add_cstring(name);
-    return stream.add_header('D');
+    return (new bits.Encoder('D'))
+      .push_raw_string(type)
+      .push_cstring(name);
   },
   Execute: function (name, max_rows) {
-    var stream = []
-      .add_cstring(name)
-      .add_int32(max_rows);
-    return stream.add_header('E');
+    return (new bits.Encoder('E'))
+      .push_cstring(name)
+      .push_int32(max_rows);
   },
   Flush: function () {
-    return [].add_header('H');
+    return new bits.Encoder('H');
   },
   FunctionCall: function () {
     // TODO: implement
   },
   Parse: function (name, query, var_types) {
-    var stream = []
-      .add_cstring(name)
-      .add_cstring(query)
-      .add_int16(var_types.length);
+    var builder = (new bits.Encoder('P'))
+      .push_cstring(name)
+      .push_cstring(query)
+      .push_int16(var_types.length);
     var_types.each(function (var_type) {
-      stream.add_int32(var_type);
+      stream.push_int32(var_type);
     });
-    return stream.add_header('P');
+    return builder;
   },
   PasswordMessage: function (password) {
-    return "".add_cstring(password).add_header('p');
+    return (new bits.Encoder('p'))
+      .push_cstring(password);
   },
   Query: function (query) {
-    return "".add_cstring(query).add_header('Q');
+    return (new bits.Encoder('Q'))
+      .push_cstring(query);
   },
   SSLRequest: function () {
-    return "".add_int32(0x4D2162F).add_header();
+    return (new bits.Encoder())
+      .push_int32(0x4D2162F);
   },
   StartupMessage: function (options) {
     // Protocol version number 3
-    return ("".add_int32(0x30000) +
-      options.map_pairs(function (key, value) {
-        return "".add_cstring(key).add_cstring(value);
-      }).join("") + "0").add_header();
+    return (new bits.Encoder())
+      .push_int32(0x30000)
+      .push_hash(options);
   },
   Sync: function () {
-    return [].add_header('S');
+    return new bits.Encoder('S');
   },
   Terminate: function () {
-    return [].add_header('X');
+    return new bits.Encoder('X');
   }
 };
 
 // Parse response streams from the server
 function parse_response(code, stream) {
+  var input = new bits.Decoder(stream);
   var type, args;
   args = [];
   switch (code) {
   case 'R':
-    switch (stream.parse_int32()[1]) {
+    switch (stream.shift_int32()) {
     case 0:
       type = "AuthenticationOk";
       break;
@@ -101,11 +87,11 @@ function parse_response(code, stream) {
       break;
     case 4:
       type = "AuthenticationCryptPassword";
-      args = stream.substr(4).parse(["raw_string", 2])[1];
+      args = [stream.shift_raw_string(2)];
       break;
     case 5:
       type = "AuthenticationMD5Password";
-      args = stream.substr(4).parse(["raw_string", 4])[1];
+      args = [stream.shift_raw_string(4)];
       break;
     case 6:
       type = "AuthenticationSCMCredential";
@@ -122,62 +108,56 @@ function parse_response(code, stream) {
   case 'E':
     type = "ErrorResponse";
     args = [{}];
-    stream.parse("multi_cstring")[1][0].forEach(function (field) {
+    stream.shift_multi_cstring().forEach(function (field) {
       args[0][field[0]] = field.substr(1);
     });
     break;
   case 'S':
     type = "ParameterStatus";
-    args = stream.parse("cstring", "cstring")[1];
+    args = [stream.shift_cstring(), stream.shift_cstring()];
     break;
   case 'K':
     type = "BackendKeyData";
-    args = stream.parse("int32", "int32")[1];
+    args = [stream.shift_int32(), stream.shift_int32()];
     break;
   case 'Z':
     type = "ReadyForQuery";
-    args = stream.parse(["raw_string", 1])[1];
+    args = [stream.shift_raw_string(1)];
     break;
   case 'T':
     type = "RowDescription";
-    var num_fields = stream.parse_int16()[1];
-    stream = stream.substr(2);
+    var num_fields = stream.shift_int16();
     var row = [];
     for (var i = 0; i < num_fields; i += 1) {
-      var parts = stream.parse("cstring", "int32", "int16", "int32", "int16", "int32", "int16");
       row.push({
-        field: parts[1][0],
-        table_id: parts[1][1],
-        column_id: parts[1][2],
-        type_id: parts[1][3],
-        type_size: parts[1][4],
-        type_modifier: parts[1][5],
-        format_code: parts[1][6]
+        field: stream.shift_cstring(),
+        table_id: stream.shift_int32(),
+        column_id: stream.shift_int16(),
+        type_id: stream.shift_int32(),
+        type_size: stream.shift_int16(),
+        type_modifier: stream.shift_int32(),
+        format_code: stream.shift_int16()
       });
-      stream = stream.substr(parts[0]);
     }
     args = [row];
     break;
   case 'D':
     type = "DataRow";
     var data = [];
-    var num_cols = stream.parse_int16()[1];
-    stream = stream.substr(2);
+    var num_cols = stream.shift_int16();
     for (i = 0; i < num_cols; i += 1) {
-      var size = stream.parse_int32()[1];
-      stream = stream.substr(4);
+      var size = stream.shift_int32();
       if (size === -1) {
         data.push(null);
       } else {
-        data.push(stream.parse_raw_string(size)[1]);
-        stream = stream.substr(size);
+        data.push(stream.shift_raw_string(size));
       }
     }
     args = [data];
     break;
   case 'C':
     type = "CommandComplete";
-    args = stream.parse("cstring")[1];
+    args = [stream.shift_cstring()];
     break;
   }
   if (!type) {
@@ -205,7 +185,7 @@ exports.Connection = function (database, username, password, port) {
 
   // Sends a message to the postgres server
   function sendMessage(type, args) {
-    var stream = formatter[type].apply(this, args);
+    var stream = (formatter[type].apply(this, args)).toString();
     if (exports.DEBUG > 0) {
       sys.debug("Sending " + type + ": " + JSON.stringify(args));
       if (exports.DEBUG > 2) {
@@ -221,16 +201,15 @@ exports.Connection = function (database, username, password, port) {
     sendMessage('StartupMessage', [{user: username, database: database}]);
   });
   connection.addListener("receive", function (data) {
-
+    var input = new bits.Decoder(data);
     if (exports.DEBUG > 2) {
       sys.debug("<-" + JSON.stringify(data));
     }
 
-    while (data.length > 0) {
-      var code = data[0];
-      var len = data.substr(1, 4).parse_int32()[1];
-      var stream = data.substr(5, len - 4);
-      data = data.substr(len + 1);
+    while (input.data.length > 0) {
+      var code = input.shift_code();
+      var len = input.shift_int32();
+      var stream = new bits.Decoder(input.shift_raw_string(len - 4));
       if (exports.DEBUG > 1) {
         sys.debug("stream: " + code + " " + JSON.stringify(stream));
       }