Support "flat" encoding, and "packed-flat" too.

src/node-capnp/capnp-test.js

@@ -27,14 +27,20 @@ var spawn = require("child_process").spawn;
 
 var goldenBinary;
 var goldenPackedBinary;
+var goldenFlatBinary;
+var goldenPackedFlatBinary;
 try {
   // Works in Ekam build.
   goldenBinary = fs.readFileSync("node-capnp/testdata/binary");
   goldenPackedBinary = fs.readFileSync("node-capnp/testdata/packedbinary");
+  goldenFlatBinary = fs.readFileSync("node-capnp/testdata/flat");
+  goldenPackedFlatBinary = fs.readFileSync("node-capnp/testdata/packedflat");
 } catch (ex) {
   // Works in npm build.
   goldenBinary = fs.readFileSync("src/node-capnp/testdata/binary");
   goldenPackedBinary = fs.readFileSync("src/node-capnp/testdata/packedbinary");
+  goldenFlatBinary = fs.readFileSync("src/node-capnp/testdata/flat");
+  goldenPackedFlatBinary = fs.readFileSync("src/node-capnp/testdata/packedflat");
 }
 
 var test = require("./test.capnp");
@@ -64,10 +70,22 @@ var roundTrippedPacked = capnp.serializePacked(test.TestAllTypes, parsedPacked);
 assert.equal(goldenPackedBinary.length, roundTrippedPacked.length, "Round trip changed size?");
 assert.equal(goldenPackedBinary.toString("base64"), roundTrippedPacked.toString("base64"), "Round trip lost data?");
 
+var parsedFlat = capnp.parse(test.TestAllTypes, goldenFlatBinary, {flat: true});
+var roundTrippedFlat = capnp.serialize(test.TestAllTypes, parsedFlat, {flat: true});
+assert.equal(goldenFlatBinary.length, roundTrippedFlat.length, "Round trip changed size?");
+assert.equal(goldenFlatBinary.toString("base64"), roundTrippedFlat.toString("base64"), "Round trip lost data?");
+
+var parsedPackedFlat = capnp.parse(test.TestAllTypes, goldenPackedFlatBinary, {packed: true, flat: true});
+var roundTrippedPackedFlat = capnp.serialize(test.TestAllTypes, parsedPackedFlat, {packed: true, flat: true});
+assert.equal(goldenPackedFlatBinary.length, roundTrippedPackedFlat.length, "Round trip changed size?");
+assert.equal(goldenPackedFlatBinary.toString("base64"), roundTrippedPackedFlat.toString("base64"), "Round trip lost data?");
+
 // TODO(someday): do a more thorough deep equality comparison of parsed and parsedPacked
 var keys = ["voidField", "boolField", "int8Field", "int16Field", "int32Field", "int64Field"];
 for (var key in keys) {
   assert.equal(parsed[key], parsedPacked[key]);
+  assert.equal(parsed[key], parsedFlat[key]);
+  assert.equal(parsed[key], parsedPackedFlat[key]);
 }
 
 // =======================================================================================

src/node-capnp/capnp.cc

@@ -1183,12 +1183,24 @@ struct StructReader {
       : message(data), root(message.getRoot<capnp::DynamicStruct>(schema)) {}
 };
 
+struct FlatStructReader {
+  kj::ArrayPtr<const capnp::word> segments[1];
+  capnp::SegmentArrayMessageReader message;
+  capnp::DynamicStruct::Reader root;
+
+  FlatStructReader(kj::ArrayPtr<const capnp::word> data, capnp::StructSchema schema)
+      : segments{data}, message(segments),
+        root(message.getRoot<capnp::DynamicStruct>(schema)) {}
+};
+
 struct PackedStructReader {
+  kj::ArrayInputStream inputStream;
   capnp::PackedMessageReader message;
   capnp::DynamicStruct::Reader root;
 
-  PackedStructReader(kj::BufferedInputStream& inputStream, capnp::StructSchema schema)
-      : message(inputStream), root(message.getRoot<capnp::DynamicStruct>(schema)) {}
+  PackedStructReader(kj::ArrayPtr<const byte> bytes, capnp::StructSchema schema)
+      : inputStream(bytes), message(inputStream),
+        root(message.getRoot<capnp::DynamicStruct>(schema)) {}
 };
 
 struct ServerRequest {
@@ -1218,6 +1230,8 @@ kj::Maybe<capnp::DynamicStruct::Reader> unwrapReader(v8::Handle<v8::Value> handl
     return reader->root;
   } else KJ_IF_MAYBE(reader, Wrapper::tryUnwrap<PackedStructReader>(handle)) {
     return reader->root;
+  } else KJ_IF_MAYBE(reader, Wrapper::tryUnwrap<FlatStructReader>(handle)) {
+    return reader->root;
   } else KJ_IF_MAYBE(request, Wrapper::tryUnwrap<ServerRequest>(handle)) {
     return request->params;
   } else KJ_IF_MAYBE(builder, unwrapBuilder(handle)) {
@@ -1776,7 +1790,7 @@ v8::Handle<v8::Value> toJsParams(const v8::Arguments& args) {
 // -----------------------------------------------------------------------------
 
 v8::Handle<v8::Value> fromBytes(const v8::Arguments& args) {
-  // fromBytes(buffer, schema) -> reader
+  // fromBytes(buffer, schema, options) -> reader
 
   KJV8_UNWRAP(CapnpContext, context, args.Data());
 
@@ -1789,72 +1803,92 @@ v8::Handle<v8::Value> fromBytes(const v8::Arguments& args) {
     return emptyHandle;
   }
 
+  if (!args[2]->IsObject()) {
+    KJV8_TYPE_ERROR(obj, v8::Object);
+  }
+
   return liftKj([&]() -> v8::Handle<v8::Value> {
-    kj::ArrayPtr<const capnp::word> words;
-    if (reinterpret_cast<uintptr_t>(buffer.begin()) % sizeof(capnp::word) != 0) {
-      // Array is not aligned.  We have to make a copy.  :(
-      auto array = kj::heapArray<capnp::word>(buffer.size() / sizeof(capnp::word));
-      memcpy(array.begin(), buffer.begin(), buffer.size());
-      words = array;
-      bufferHandle = context.wrapper.wrapCopy(kj::mv(array));
+    auto options = args[2].As<v8::Object>();
+    bool packed = options->Get(v8::String::NewSymbol("packed"))->BooleanValue();
+    bool flat = options->Get(v8::String::NewSymbol("flat"))->BooleanValue();
+
+    v8::Local<v8::Object> wrapper;
+
+    if (packed) {
+      if (flat) {
+        auto bytes = kj::arrayPtr(buffer.begin(), buffer.size());
+        auto words = kj::heapArray<capnp::word>(capnp::computeUnpackedSizeInWords(bytes));
+        kj::ArrayInputStream input(bytes);
+        capnp::_::PackedInputStream unpacker(input);
+        unpacker.read(words.asBytes().begin(), words.asBytes().size());
+        wrapper = context.wrapper.wrap(new FlatStructReader(words, schema.asStruct()));
+        bufferHandle = context.wrapper.wrapCopy(kj::mv(words));
+      } else {
+        wrapper = context.wrapper.wrap(new PackedStructReader(
+            kj::arrayPtr(buffer.begin(), buffer.size()), schema.asStruct()));
+      }
     } else {
-      // Yay, array is aligned.
-      words = kj::arrayPtr(reinterpret_cast<const capnp::word*>(buffer.begin()),
-                           buffer.size() / sizeof(capnp::word));
+      kj::ArrayPtr<const capnp::word> words;
+      if (reinterpret_cast<uintptr_t>(buffer.begin()) % sizeof(capnp::word) != 0) {
+        // Array is not aligned.  We have to make a copy.  :(
+        auto array = kj::heapArray<capnp::word>(buffer.size() / sizeof(capnp::word));
+        memcpy(array.begin(), buffer.begin(), buffer.size());
+        words = array;
+        bufferHandle = context.wrapper.wrapCopy(kj::mv(array));
+      } else {
+        // Yay, array is aligned.
+        words = kj::arrayPtr(reinterpret_cast<const capnp::word*>(buffer.begin()),
+                             buffer.size() / sizeof(capnp::word));
+      }
+
+      if (flat) {
+        wrapper = context.wrapper.wrap(new FlatStructReader(words, schema.asStruct()));
+      } else {
+        wrapper = context.wrapper.wrap(new StructReader(words, schema.asStruct()));
+      }
     }
 
-    auto wrapper = context.wrapper.wrap(new StructReader(words, schema.asStruct()));
-    wrapper->SetHiddenValue(v8::String::NewSymbol("buffer"), bufferHandle);
-    return wrapper;
-  });
-}
-
-v8::Handle<v8::Value> fromBytesPacked(const v8::Arguments& args) {
-  // fromBytesPacked(buffer, schema) -> reader
-  KJV8_UNWRAP(CapnpContext, context, args.Data());
-
-  v8::Handle<v8::Value> bufferHandle = args[0];
-  KJV8_UNWRAP_BUFFER(buffer, bufferHandle);
-
-  KJV8_UNWRAP(capnp::Schema, schema, args[1]);
-  if (!schema.getProto().isStruct()) {
-    KJV8_TYPE_ERROR(schema, capnp::StructSchema);
-    return emptyHandle;
-  }
-
-  return liftKj([&]() -> v8::Handle<v8::Value> {
-    kj::ArrayInputStream inputStream(kj::arrayPtr(buffer.begin(), buffer.size()));
-    auto wrapper = context.wrapper.wrap(new PackedStructReader(inputStream, schema.asStruct()));
     wrapper->SetHiddenValue(v8::String::NewSymbol("buffer"), bufferHandle);
     return wrapper;
   });
 }
 
 v8::Handle<v8::Value> toBytes(const v8::Arguments& args) {
-  // toBytes(builder) -> buffer
+  // toBytes(builder, options) -> buffer
 
   KJV8_UNWRAP(StructBuilder, builder, args[0]);
+  if (!args[1]->IsObject()) {
+    KJV8_TYPE_ERROR(obj, v8::Object);
+  }
 
   return liftKj([&]() -> v8::Handle<v8::Value> {
-    return wrapBuffer(capnp::messageToFlatArray(builder.message));
-  });
-}
+    auto options = args[1].As<v8::Object>();
+    bool packed = options->Get(v8::String::NewSymbol("packed"))->BooleanValue();
+    bool flat = options->Get(v8::String::NewSymbol("flat"))->BooleanValue();
 
-v8::Handle<v8::Value> toBytesPacked(const v8::Arguments& args) {
-  // toBytes(builder) -> buffer
+    if (flat) {
+      // Write whole message to flat array.
+      auto root = builder.root.asReader();
+      auto words = kj::heapArray<capnp::word>(root.totalSize().wordCount + 1);
+      memset(words.asBytes().begin(), 0, words.asBytes().size());
+      capnp::copyToUnchecked(root, words);
 
-  KJV8_UNWRAP(StructBuilder, builder, args[0]);
-
-  return liftKj([&]() -> v8::Handle<v8::Value> {
-    auto unpackedSize = computeSerializedSizeInWords(builder.message) * sizeof(capnp::word);
-    // Packed encoding can add 2 bytes of overhead per 2k segment reached.
-    auto packedOverhead = ((unpackedSize + 2047) / 2048) * 2;
-    auto packedSizeUpperBound = unpackedSize + packedOverhead;
-    auto oversizedBuf = kj::heapArray<byte>(packedSizeUpperBound);
-    kj::ArrayOutputStream outputStream(oversizedBuf);
-    capnp::writePackedMessage(outputStream, builder.message);
-    auto packedMessage = heapArray(outputStream.getArray());
-    return wrapBuffer(kj::mv(packedMessage));
+      if (packed) {
+        kj::VectorOutputStream output;
+        capnp::_::PackedOutputStream packer(output);
+        packer.write(words.asBytes().begin(), words.asBytes().size());
+        return wrapBuffer(kj::heapArray(output.getArray()));
+      } else {
+        return wrapBuffer(kj::mv(words));
+      }
+    } else if (packed) {
+      kj::VectorOutputStream output;
+      capnp::writePackedMessage(output, builder.message);
+      auto packedMessage = heapArray(output.getArray());
+      return wrapBuffer(kj::mv(packedMessage));
+    } else {
+      return wrapBuffer(capnp::messageToFlatArray(builder.message));
+    }
   });
 }
 
@@ -2548,9 +2582,7 @@ void init(v8::Handle<v8::Object> exports) {
     mapFunction("toJs", toJs);
     mapFunction("toJsParams", toJsParams);
     mapFunction("fromBytes", fromBytes);
-    mapFunction("fromBytesPacked", fromBytesPacked);
     mapFunction("toBytes", toBytes);
-    mapFunction("toBytesPacked", toBytesPacked);
     mapFunction("connect", connect);
     mapFunction("disconnect", disconnect);
     mapFunction("restore", restore);

src/node-capnp/capnp.js

@@ -269,35 +269,34 @@ exports.connect = function (addr, bootstrapCap) {
   return new Connection(v8capnp.connect(addr, bootstrapCap));
 }
 
-exports.parse = function (schema, buffer) {
-  var reader = v8capnp.fromBytes(buffer, schema);
+exports.parse = function (schema, buffer, options) {
+  var reader = v8capnp.fromBytes(buffer, schema, options || {});
   return v8capnp.toJs(reader, Capability);
 }
 
 exports.parsePacked = function (schema, buffer) {
-  var reader = v8capnp.fromBytesPacked(buffer, schema);
-  return v8capnp.toJs(reader, Capability);
+  // TODO(cleanup): Remove in favor of options.
+  return exports.parse(schema, buffer, {packed: true});
 }
 
-exports.serialize = function (schema, value) {
+exports.serialize = function (schema, value, options) {
   var builder = v8capnp.newBuilder(schema);
   v8capnp.fromJs(builder, value, LocalCapWrapper);
-  return v8capnp.toBytes(builder);
+  return v8capnp.toBytes(builder, options || {});
 }
 
 exports.serializePacked = function (schema, value) {
-  var builder = v8capnp.newBuilder(schema);
-  v8capnp.fromJs(builder, value, LocalCapWrapper);
-  return v8capnp.toBytesPacked(builder);
+  // TODO(cleanup): Remove in favor of options.
+  return exports.serialize(schema, value, {packed: true});
 }
 
 exports.Capability = Capability;
 
-exports.bytesToPreorder = function(schema, buf) {
+exports.bytesToPreorder = function(schema, buf, options) {
   // Parse, copy, serialize, to get a preorder traversal.
-  var reader = v8capnp.fromBytes(buf, schema);
+  var reader = v8capnp.fromBytes(buf, schema, options || {});
   var builder = v8capnp.copyBuilder(reader);
-  return v8capnp.toBytes(builder);
+  return v8capnp.toBytes(builder, options || {});
 }
 
 var verboseDebugLogging = false;