Changed low-level metadata API to be more reasonable

src/node/ext/call.cc

@@ -33,6 +33,7 @@
 
 #include <node.h>
 
+#include "grpc/support/log.h"
 #include "grpc/grpc.h"
 #include "grpc/support/time.h"
 #include "byte_buffer.h"
@@ -173,31 +174,43 @@ NAN_METHOD(Call::AddMetadata) {
     return NanThrowTypeError("addMetadata can only be called on Call objects");
   }
   Call *call = ObjectWrap::Unwrap<Call>(args.This());
-  for (int i = 0; !args[i]->IsUndefined(); i++) {
+  if (!args[0]->IsObject()) {
-    if (!args[i]->IsObject()) {
+    return NanThrowTypeError("addMetadata's first argument must be an object");
+  }
+  Handle<Object> metadata = args[0]->ToObject();
+  Handle<Array> keys(metadata->GetOwnPropertyNames());
+  for (unsigned int i = 0; i < keys->Length(); i++) {
+    Handle<String> current_key(keys->Get(i)->ToString());
+    if (!metadata->Get(current_key)->IsArray()) {
       return NanThrowTypeError(
-          "addMetadata arguments must be objects with key and value");
+          "addMetadata's first argument's values must be arrays");
     }
-    Handle<Object> item = args[i]->ToObject();
+    NanUtf8String utf8_key(current_key);
-    Handle<Value> key = item->Get(NanNew("key"));
+    Handle<Array> values = Local<Array>::Cast(metadata->Get(current_key));
-    if (!key->IsString()) {
+    for (unsigned int j = 0; j < values->Length(); j++) {
-      return NanThrowTypeError(
+      Handle<Value> value = values->Get(j);
-          "objects passed to addMetadata must have key->string");
+      grpc_metadata metadata;
-    }
+      grpc_call_error error;
-    Handle<Value> value = item->Get(NanNew("value"));
+      metadata.key = *utf8_key;
-    if (!Buffer::HasInstance(value)) {
+      if (Buffer::HasInstance(value)) {
-      return NanThrowTypeError(
+        metadata.value = Buffer::Data(value);
-          "objects passed to addMetadata must have value->Buffer");
+        metadata.value_length = Buffer::Length(value);
-    }
+        error = grpc_call_add_metadata(call->wrapped_call, &metadata, 0);
-    grpc_metadata metadata;
+      } else if (value->IsString()) {
-    NanUtf8String utf8_key(key);
+        Handle<String> string_value = value->ToString();
-    metadata.key = *utf8_key;
+        NanUtf8String utf8_value(string_value);
-    metadata.value = Buffer::Data(value);
+        metadata.value = *utf8_value;
-    metadata.value_length = Buffer::Length(value);
+        metadata.value_length = string_value->Length();
-    grpc_call_error error =
+        gpr_log(GPR_DEBUG, "adding metadata: %s, %s, %d", metadata.key,
-        grpc_call_add_metadata(call->wrapped_call, &metadata, 0);
+                metadata.value, metadata.value_length);
-    if (error != GRPC_CALL_OK) {
+        error = grpc_call_add_metadata(call->wrapped_call, &metadata, 0);
-      return NanThrowError("addMetadata failed", error);
+      } else {
+        return NanThrowTypeError(
+            "addMetadata values must be strings or buffers");
+      }
+      if (error != GRPC_CALL_OK) {
+        return NanThrowError("addMetadata failed", error);
+      }
     }
   }
   NanReturnUndefined();

src/node/ext/event.cc

@@ -31,6 +31,8 @@
  *
  */
 
+#include <map>
+
 #include <node.h>
 #include <nan.h>
 #include "grpc/grpc.h"
@@ -43,6 +45,7 @@
 namespace grpc {
 namespace node {
 
+using ::node::Buffer;
 using v8::Array;
 using v8::Date;
 using v8::Handle;
@@ -53,6 +56,36 @@ using v8::Persistent;
 using v8::String;
 using v8::Value;
 
+Handle<Value> ParseMetadata(grpc_metadata *metadata_elements, size_t length) {
+  NanEscapableScope();
+  std::map<char*, size_t> size_map;
+  std::map<char*, size_t> index_map;
+
+  for (unsigned int i = 0; i < length; i++) {
+    char *key = metadata_elements[i].key;
+    if (size_map.count(key)) {
+      size_map[key] += 1;
+    }
+    index_map[key] = 0;
+  }
+  Handle<Object> metadata_object = NanNew<Object>();
+  for (unsigned int i = 0; i < length; i++) {
+    grpc_metadata* elem = &metadata_elements[i];
+    Handle<String> key_string = String::New(elem->key);
+    Handle<Array> array;
+    if (metadata_object->Has(key_string)) {
+      array = Handle<Array>::Cast(metadata_object->Get(key_string));
+    } else {
+      array = NanNew<Array>(size_map[elem->key]);
+      metadata_object->Set(key_string, array);
+    }
+    array->Set(index_map[elem->key],
+               NanNewBufferHandle(elem->value, elem->value_length));
+    index_map[elem->key] += 1;
+  }
+  return NanEscapeScope(metadata_object);
+}
+
 Handle<Value> GetEventData(grpc_event *event) {
   NanEscapableScope();
   size_t count;
@@ -72,18 +105,7 @@ Handle<Value> GetEventData(grpc_event *event) {
     case GRPC_CLIENT_METADATA_READ:
       count = event->data.client_metadata_read.count;
       items = event->data.client_metadata_read.elements;
-      metadata = NanNew<Array>(static_cast<int>(count));
+      return NanEscapeScope(ParseMetadata(items, count));
-      for (unsigned int i = 0; i < count; i++) {
-        Handle<Object> item_obj = NanNew<Object>();
-        item_obj->Set(NanNew<String, const char *>("key"),
-                      NanNew<String, char *>(items[i].key));
-        item_obj->Set(
-            NanNew<String, const char *>("value"),
-            NanNew<String, char *>(items[i].value,
-                                   static_cast<int>(items[i].value_length)));
-        metadata->Set(i, item_obj);
-      }
-      return NanEscapeScope(metadata);
     case GRPC_FINISHED:
       status = NanNew<Object>();
       status->Set(NanNew("code"), NanNew<Number>(event->data.finished.status));
@@ -93,18 +115,7 @@ Handle<Value> GetEventData(grpc_event *event) {
       }
       count = event->data.finished.metadata_count;
       items = event->data.finished.metadata_elements;
-      metadata = NanNew<Array>(static_cast<int>(count));
+      status->Set(NanNew("metadata"), ParseMetadata(items, count));
-      for (unsigned int i = 0; i < count; i++) {
-        Handle<Object> item_obj = NanNew<Object>();
-        item_obj->Set(NanNew<String, const char *>("key"),
-                      NanNew<String, char *>(items[i].key));
-        item_obj->Set(
-            NanNew<String, const char *>("value"),
-            NanNew<String, char *>(items[i].value,
-                                   static_cast<int>(items[i].value_length)));
-        metadata->Set(i, item_obj);
-      }
-      status->Set(NanNew("metadata"), metadata);
       return NanEscapeScope(status);
     case GRPC_SERVER_RPC_NEW:
       rpc_new = NanNew<Object>();
@@ -133,7 +144,7 @@ Handle<Value> GetEventData(grpc_event *event) {
                                    static_cast<int>(items[i].value_length)));
         metadata->Set(i, item_obj);
       }
-      rpc_new->Set(NanNew<String, const char *>("metadata"), metadata);
+      rpc_new->Set(NanNew("metadata"), ParseMetadata(items, count));
       return NanEscapeScope(rpc_new);
     default:
       return NanEscapeScope(NanNull());

src/node/test/call_test.js

@@ -99,24 +99,30 @@ describe('call', function() {
     });
   });
   describe('addMetadata', function() {
-    it('should succeed with objects containing keys and values', function() {
+    it('should succeed with a map from strings to string arrays', function() {
       var call = new grpc.Call(channel, 'method', getDeadline(1));
       assert.doesNotThrow(function() {
-        call.addMetadata();
+        call.addMetadata({'key': ['value']});
+      });
+      assert.doesNotThrow(function() {
+        call.addMetadata({'key1': ['value1'], 'key2': ['value2']});
       });
+    });
+    it('should succeed with a map from strings to buffer arrays', function() {
+      var call = new grpc.Call(channel, 'method', getDeadline(1));
       assert.doesNotThrow(function() {
-        call.addMetadata({'key' : 'key',
+        call.addMetadata({'key': [new Buffer('value')]});
-                          'value' : new Buffer('value')});
       });
       assert.doesNotThrow(function() {
-        call.addMetadata({'key' : 'key1',
+        call.addMetadata({'key1': [new Buffer('value1')],
-                          'value' : new Buffer('value1')},
+                          'key2': [new Buffer('value2')]});
-                         {'key' : 'key2',
-                          'value' : new Buffer('value2')});
       });
     });
     it('should fail with other parameter types', function() {
       var call = new grpc.Call(channel, 'method', getDeadline(1));
+      assert.throws(function() {
+        call.addMetadata();
+      });
       assert.throws(function() {
         call.addMetadata(null);
       }, TypeError);
@@ -133,7 +139,7 @@ describe('call', function() {
                   function() {done();},
                   0);
       assert.throws(function() {
-        call.addMetadata({'key' : 'key', 'value' : new Buffer('value') });
+        call.addMetadata({'key': ['value']});
       }, function(err) {
         return err.code === grpc.callError.ALREADY_INVOKED;
       });

src/node/test/end_to_end_test.js

@@ -68,18 +68,61 @@ describe('end-to-end', function() {
     server.shutdown();
   });
   it('should start and end a request without error', function(complete) {
-    var done = multiDone(function() {
+    var done = multiDone(complete, 2);
-      complete();
-    }, 2);
     var deadline = new Date();
     deadline.setSeconds(deadline.getSeconds() + 3);
     var status_text = 'xyz';
     var call = new grpc.Call(channel,
                              'dummy_method',
                              deadline);
-      call.invoke(function(event) {
+    call.invoke(function(event) {
+      assert.strictEqual(event.type,
+                         grpc.completionType.CLIENT_METADATA_READ);
+    },function(event) {
+      assert.strictEqual(event.type, grpc.completionType.FINISHED);
+      var status = event.data;
+      assert.strictEqual(status.code, grpc.status.OK);
+      assert.strictEqual(status.details, status_text);
+      done();
+    }, 0);
+
+    server.requestCall(function(event) {
+      assert.strictEqual(event.type, grpc.completionType.SERVER_RPC_NEW);
+      var server_call = event.call;
+      assert.notEqual(server_call, null);
+      server_call.serverAccept(function(event) {
+        assert.strictEqual(event.type, grpc.completionType.FINISHED);
+      }, 0);
+      server_call.serverEndInitialMetadata(0);
+      server_call.startWriteStatus(
+          grpc.status.OK,
+          status_text,
+          function(event) {
+            assert.strictEqual(event.type,
+                               grpc.completionType.FINISH_ACCEPTED);
+            assert.strictEqual(event.data, grpc.opError.OK);
+            done();
+          });
+    });
+    call.writesDone(function(event) {
+      assert.strictEqual(event.type,
+                         grpc.completionType.FINISH_ACCEPTED);
+      assert.strictEqual(event.data, grpc.opError.OK);
+    });
+  });
+  it('should successfully send and receive metadata', function(complete) {
+    var done = multiDone(complete, 2);
+    var deadline = new Date();
+    deadline.setSeconds(deadline.getSeconds() + 3);
+    var status_text = 'xyz';
+    var call = new grpc.Call(channel,
+                             'dummy_method',
+                             deadline);
+    call.addMetadata({'client_key': ['client_value']});
+    call.invoke(function(event) {
       assert.strictEqual(event.type,
                          grpc.completionType.CLIENT_METADATA_READ);
+      assert.strictEqual(event.data.server_key[0].toString(), 'server_value');
     },function(event) {
       assert.strictEqual(event.type, grpc.completionType.FINISHED);
       var status = event.data;
@@ -90,11 +133,14 @@ describe('end-to-end', function() {
 
     server.requestCall(function(event) {
       assert.strictEqual(event.type, grpc.completionType.SERVER_RPC_NEW);
+      assert.strictEqual(event.data.metadata.client_key[0].toString(),
+                         'client_value');
       var server_call = event.call;
       assert.notEqual(server_call, null);
       server_call.serverAccept(function(event) {
         assert.strictEqual(event.type, grpc.completionType.FINISHED);
       }, 0);
+      server_call.addMetadata({'server_key': ['server_value']});
       server_call.serverEndInitialMetadata(0);
       server_call.startWriteStatus(
           grpc.status.OK,
@@ -115,10 +161,7 @@ describe('end-to-end', function() {
   it('should send and receive data without error', function(complete) {
     var req_text = 'client_request';
     var reply_text = 'server_response';
-    var done = multiDone(function() {
+    var done = multiDone(complete, 6);
-      complete();
-      server.shutdown();
-    }, 6);
     var deadline = new Date();
     deadline.setSeconds(deadline.getSeconds() + 3);
     var status_text = 'success';

src/node/test/server_test.js

@@ -75,6 +75,9 @@ describe('echo server', function() {
 
     channel = new grpc.Channel('localhost:' + port_num);
   });
+  after(function() {
+    server.shutdown();
+  });
   it('should echo inputs as responses', function(done) {
     done = multiDone(done, 4);
 
@@ -95,7 +98,6 @@ describe('echo server', function() {
       var status = event.data;
       assert.strictEqual(status.code, grpc.status.OK);
       assert.strictEqual(status.details, status_text);
-      server.shutdown();
       done();
     }, 0);
     call.startWrite(