Add parent call propagation API to Node library

6 files changed, 244 insertions, 5 deletions

diff --git a/src/node/ext/call.cc b/src/node/ext/call.cc
index c5c8313385..5e187607cc 100644
--- a/src/node/ext/call.cc
+++ b/src/node/ext/call.cc
@@ -502,6 +502,22 @@ NAN_METHOD(Call::New) {
         return NanThrowTypeError(
             "Call's third argument must be a date or a number");
       }
+      // These arguments are at the end because they are optional
+      grpc_call *parent_call = NULL;
+      if (Call::HasInstance(args[4])) {
+        Call *parent_obj = ObjectWrap::Unwrap<Call>(args[4]->ToObject());
+        parent_call = parent_obj->wrapped_call;
+      } else if (!(args[4]->IsUndefined() || args[4]->IsNull())) {
+        return NanThrowTypeError(
+            "Call's fifth argument must be another call, if provided");
+      }
+      gpr_uint32 propagate_flags = GRPC_PROPAGATE_DEFAULTS;
+      if (args[5]->IsUint32()) {
+        propagate_flags = args[5]->Uint32Value();
+      } else if (!(args[5]->IsUndefined() || args[5]->IsNull())) {
+        return NanThrowTypeError(
+            "Call's fifth argument must be propagate flags, if provided");
+      }
       Handle<Object> channel_object = args[0]->ToObject();
       Channel *channel = ObjectWrap::Unwrap<Channel>(channel_object);
       if (channel->GetWrappedChannel() == NULL) {
@@ -514,12 +530,12 @@ NAN_METHOD(Call::New) {
       if (args[3]->IsString()) {
         NanUtf8String host_override(args[3]);
         wrapped_call = grpc_channel_create_call(
-            wrapped_channel, NULL, GRPC_PROPAGATE_DEFAULTS,
+            wrapped_channel, parent_call, propagate_flags,
             CompletionQueueAsyncWorker::GetQueue(), *method,
             *host_override, MillisecondsToTimespec(deadline));
       } else if (args[3]->IsUndefined() || args[3]->IsNull()) {
         wrapped_call = grpc_channel_create_call(
-            wrapped_channel, NULL, GRPC_PROPAGATE_DEFAULTS,
+            wrapped_channel, parent_call, propagate_flags,
             CompletionQueueAsyncWorker::GetQueue(), *method,
             NULL, MillisecondsToTimespec(deadline));
       } else {
diff --git a/src/node/ext/node_grpc.cc b/src/node/ext/node_grpc.cc
index 4e31cbaa27..6c9bc8735d 100644
--- a/src/node/ext/node_grpc.cc
+++ b/src/node/ext/node_grpc.cc
@@ -159,12 +159,32 @@ void InitOpTypeConstants(Handle<Object> exports) {
   op_type->Set(NanNew("RECV_CLOSE_ON_SERVER"), RECV_CLOSE_ON_SERVER);
 }
 
+void InitPropagateConstants(Handle<Object> exports) {
+  NanScope();
+  Handle<Object> propagate = NanNew<Object>();
+  exports->Set(NanNew("propagate"), propagate);
+  Handle<Value> DEADLINE(NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_DEADLINE));
+  propagate->Set(NanNew("DEADLINE"), DEADLINE);
+  Handle<Value> CENSUS_STATS_CONTEXT(
+      NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_CENSUS_STATS_CONTEXT));
+  propagate->Set(NanNew("CENSUS_STATS_CONTEXT"), CENSUS_STATS_CONTEXT);
+  Handle<Value> CENSUS_TRACING_CONTEXT(
+      NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_CENSUS_TRACING_CONTEXT));
+  propagate->Set(NanNew("CENSUS_TRACING_CONTEXT"), CENSUS_TRACING_CONTEXT);
+  Handle<Value> CANCELLATION(
+      NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_CANCELLATION));
+  propagate->Set(NanNew("CANCELLATION"), CANCELLATION);
+  Handle<Value> DEFAULTS(NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_DEFAULTS));
+  propagate->Set(NanNew("DEFAULTS"), DEFAULTS);
+}
+
 void init(Handle<Object> exports) {
   NanScope();
   grpc_init();
   InitStatusConstants(exports);
   InitCallErrorConstants(exports);
   InitOpTypeConstants(exports);
+  InitPropagateConstants(exports);
 
   grpc::node::Call::Init(exports);
   grpc::node::Channel::Init(exports);
diff --git a/src/node/src/client.js b/src/node/src/client.js
index 5cde438572..616b3969c0 100644
--- a/src/node/src/client.js
+++ b/src/node/src/client.js
@@ -216,14 +216,19 @@ ClientDuplexStream.prototype.getPeer = getPeer;
 function getCall(channel, method, options) {
   var deadline;
   var host;
+  var parent;
+  var propagate_flags;
   if (options) {
     deadline = options.deadline;
     host = options.host;
+    parent = _.get(options, 'parent.call');
+    propagate_flags = options.propagate_flags;
   }
   if (deadline === undefined) {
     deadline = Infinity;
   }
-  return new grpc.Call(channel, method, deadline, host);
+  return new grpc.Call(channel, method, deadline, host,
+                       parent, propagate_flags);
 }
 
 /**
diff --git a/src/node/src/server.js b/src/node/src/server.js
index 5c62f5990c..8b86173f08 100644
--- a/src/node/src/server.js
+++ b/src/node/src/server.js
@@ -432,6 +432,7 @@ function handleUnary(call, handler, metadata) {
   });
   emitter.metadata = metadata;
   waitForCancel(call, emitter);
+  emitter.call = call;
   var batch = {};
   batch[grpc.opType.RECV_MESSAGE] = true;
   call.startBatch(batch, function(err, result) {
diff --git a/src/node/test/constant_test.js b/src/node/test/constant_test.js
index ecc98ec443..964fc60da0 100644
--- a/src/node/test/constant_test.js
+++ b/src/node/test/constant_test.js
@@ -78,6 +78,18 @@ var callErrorNames = [
   'INVALID_FLAGS'
 ];
 
+/**
+ * List of all propagate flag names
+ * @const
+ * @type {Array.<string>}
+ */
+var propagateFlagNames = [
+  'DEADLINE',
+  'CENSUS_STATS_CONTEXT',
+  'CENSUS_TRACING_CONTEXT',
+  'CANCELLATION'
+];
+
 describe('constants', function() {
   it('should have all of the status constants', function() {
     for (var i = 0; i < statusNames.length; i++) {
@@ -91,4 +103,10 @@ describe('constants', function() {
              'call error missing: ' + callErrorNames[i]);
     }
   });
+  it('should have all of the propagate flags', function() {
+    for (var i = 0; i < propagateFlagNames.length; i++) {
+      assert(grpc.propagate.hasOwnProperty(propagateFlagNames[i]),
+             'call error missing: ' + propagateFlagNames[i]);
+    }
+  });
 });
diff --git a/src/node/test/surface_test.js b/src/node/test/surface_test.js
index dda2f8d127..b8740af74a 100644
--- a/src/node/test/surface_test.js
+++ b/src/node/test/surface_test.js
@@ -47,6 +47,27 @@ var mathService = math_proto.lookup('math.Math');
 
 var _ = require('lodash');
 
+/**
+ * This is used for testing functions with multiple asynchronous calls that
+ * can happen in different orders. This should be passed the number of async
+ * function invocations that can occur last, and each of those should call this
+ * function's return value
+ * @param {function()} done The function that should be called when a test is
+ *     complete.
+ * @param {number} count The number of calls to the resulting function if the
+ *     test passes.
+ * @return {function()} The function that should be called at the end of each
+ *     sequence of asynchronous functions.
+ */
+function multiDone(done, count) {
+  return function() {
+    count -= 1;
+    if (count <= 0) {
+      done();
+    }
+  };
+}
+
 var server_insecure_creds = grpc.ServerCredentials.createInsecure();
 
 describe('File loader', function() {
@@ -272,12 +293,14 @@ describe('Echo metadata', function() {
   });
 });
 describe('Other conditions', function() {
+  var test_service;
+  var Client;
   var client;
   var server;
   var port;
   before(function() {
     var test_proto = ProtoBuf.loadProtoFile(__dirname + '/test_service.proto');
-    var test_service = test_proto.lookup('TestService');
+    test_service = test_proto.lookup('TestService');
     server = new grpc.Server();
     server.addProtoService(test_service, {
       unary: function(call, cb) {
@@ -339,7 +362,7 @@ describe('Other conditions', function() {
       }
     });
     port = server.bind('localhost:0', server_insecure_creds);
-    var Client = surface_client.makeProtobufClientConstructor(test_service);
+    Client = surface_client.makeProtobufClientConstructor(test_service);
     client = new Client('localhost:' + port, grpc.Credentials.createInsecure());
     server.start();
   });
@@ -592,6 +615,162 @@ describe('Other conditions', function() {
       });
     });
   });
+  describe('Call propagation', function() {
+    var proxy;
+    var proxy_impl;
+    beforeEach(function() {
+      proxy = new grpc.Server();
+      proxy_impl = {
+        unary: function(call) {},
+        clientStream: function(stream) {},
+        serverStream: function(stream) {},
+        bidiStream: function(stream) {}
+      };
+    });
+    afterEach(function() {
+      console.log('Shutting down server');
+      proxy.shutdown();
+    });
+    describe('Cancellation', function() {
+      it('With a unary call', function(done) {
+        done = multiDone(done, 2);
+        proxy_impl.unary = function(parent, callback) {
+          client.unary(parent.request, function(err, value) {
+            try {
+              assert(err);
+              assert.strictEqual(err.code, grpc.status.CANCELLED);
+            } finally {
+              callback(err, value);
+              done();
+            }
+          }, null, {parent: parent});
+          call.cancel();
+        };
+        proxy.addProtoService(test_service, proxy_impl);
+        var proxy_port = proxy.bind('localhost:0', server_insecure_creds);
+        proxy.start();
+        var proxy_client = new Client('localhost:' + proxy_port,
+                                      grpc.Credentials.createInsecure());
+        var call = proxy_client.unary({}, function(err, value) {
+          done();
+        });
+      });
+      it('With a client stream call', function(done) {
+        done = multiDone(done, 2);
+        proxy_impl.clientStream = function(parent, callback) {
+          client.clientStream(function(err, value) {
+            try {
+              assert(err);
+              assert.strictEqual(err.code, grpc.status.CANCELLED);
+            } finally {
+              callback(err, value);
+              done();
+            }
+          }, null, {parent: parent});
+          call.cancel();
+        };
+        proxy.addProtoService(test_service, proxy_impl);
+        var proxy_port = proxy.bind('localhost:0', server_insecure_creds);
+        proxy.start();
+        var proxy_client = new Client('localhost:' + proxy_port,
+                                      grpc.Credentials.createInsecure());
+        var call = proxy_client.clientStream(function(err, value) {
+          done();
+        });
+      });
+      it('With a server stream call', function(done) {
+        done = multiDone(done, 2);
+        proxy_impl.serverStream = function(parent) {
+          var child = client.serverStream(parent.request, null,
+                                          {parent: parent});
+          child.on('error', function(err) {
+            assert(err);
+            assert.strictEqual(err.code, grpc.status.CANCELLED);
+            done();
+          });
+          call.cancel();
+        };
+        proxy.addProtoService(test_service, proxy_impl);
+        var proxy_port = proxy.bind('localhost:0', server_insecure_creds);
+        proxy.start();
+        var proxy_client = new Client('localhost:' + proxy_port,
+                                      grpc.Credentials.createInsecure());
+        var call = proxy_client.serverStream({});
+        call.on('error', function(err) {
+          done();
+        });
+      });
+      it('With a bidi stream call', function(done) {
+        done = multiDone(done, 2);
+        proxy_impl.bidiStream = function(parent) {
+          var child = client.bidiStream(null, {parent: parent});
+          child.on('error', function(err) {
+            assert(err);
+            assert.strictEqual(err.code, grpc.status.CANCELLED);
+            done();
+          });
+          call.cancel();
+        };
+        proxy.addProtoService(test_service, proxy_impl);
+        var proxy_port = proxy.bind('localhost:0', server_insecure_creds);
+        proxy.start();
+        var proxy_client = new Client('localhost:' + proxy_port,
+                                      grpc.Credentials.createInsecure());
+        var call = proxy_client.bidiStream();
+        call.on('error', function(err) {
+          done();
+        });
+      });
+    });
+    describe('Deadline', function() {
+      it.skip('With a client stream call', function(done) {
+        done = multiDone(done, 2);
+        proxy_impl.clientStream = function(parent, callback) {
+          client.clientStream(function(err, value) {
+            try {
+              assert(err);
+              assert.strictEqual(err.code, grpc.status.DEADLINE_EXCEEDED);
+            } finally {
+              callback(err, value);
+              done();
+            }
+          }, null, {parent: parent});
+        };
+        proxy.addProtoService(test_service, proxy_impl);
+        var proxy_port = proxy.bind('localhost:0', server_insecure_creds);
+        proxy.start();
+        var proxy_client = new Client('localhost:' + proxy_port,
+                                      grpc.Credentials.createInsecure());
+        var deadline = new Date();
+        deadline.setSeconds(deadline.getSeconds() + 1);
+        var call = proxy_client.clientStream(function(err, value) {
+          done();
+        }, null, {deadline: deadline});
+      });
+      it.skip('With a bidi stream call', function(done) {
+        done = multiDone(done, 2);
+        proxy_impl.bidiStream = function(parent) {
+          var child = client.bidiStream(null, {parent: parent});
+          child.on('error', function(err) {
+            assert(err);
+            assert.strictEqual(err.code, grpc.status.DEADLINE_EXCEEDED);
+            done();
+          });
+        };
+        proxy.addProtoService(test_service, proxy_impl);
+        var proxy_port = proxy.bind('localhost:0', server_insecure_creds);
+        proxy.start();
+        var proxy_client = new Client('localhost:' + proxy_port,
+                                      grpc.Credentials.createInsecure());
+        var deadline = new Date();
+        deadline.setSeconds(deadline.getSeconds() + 1);
+        var call = proxy_client.bidiStream(null, {deadline: deadline});
+        call.on('error', function(err) {
+          done();
+        });
+      });
+    });
+  });
 });
 describe('Cancelling surface client', function() {
   var client;