diff --git a/ext/call.cc b/ext/call.cc index 6fc1bc42..705c80ff 100644 --- a/ext/call.cc +++ b/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(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 sixth argument must be propagate flags, if provided"); + } Handle channel_object = args[0]->ToObject(); Channel *channel = ObjectWrap::Unwrap(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), NULL); } 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), NULL); } else { diff --git a/ext/node_grpc.cc b/ext/node_grpc.cc index 331ccb60..d93dafda 100644 --- a/ext/node_grpc.cc +++ b/ext/node_grpc.cc @@ -159,6 +159,25 @@ void InitOpTypeConstants(Handle exports) { op_type->Set(NanNew("RECV_CLOSE_ON_SERVER"), RECV_CLOSE_ON_SERVER); } +void InitPropagateConstants(Handle exports) { + NanScope(); + Handle propagate = NanNew(); + exports->Set(NanNew("propagate"), propagate); + Handle DEADLINE(NanNew(GRPC_PROPAGATE_DEADLINE)); + propagate->Set(NanNew("DEADLINE"), DEADLINE); + Handle CENSUS_STATS_CONTEXT( + NanNew(GRPC_PROPAGATE_CENSUS_STATS_CONTEXT)); + propagate->Set(NanNew("CENSUS_STATS_CONTEXT"), CENSUS_STATS_CONTEXT); + Handle CENSUS_TRACING_CONTEXT( + NanNew(GRPC_PROPAGATE_CENSUS_TRACING_CONTEXT)); + propagate->Set(NanNew("CENSUS_TRACING_CONTEXT"), CENSUS_TRACING_CONTEXT); + Handle CANCELLATION( + NanNew(GRPC_PROPAGATE_CANCELLATION)); + propagate->Set(NanNew("CANCELLATION"), CANCELLATION); + Handle DEFAULTS(NanNew(GRPC_PROPAGATE_DEFAULTS)); + propagate->Set(NanNew("DEFAULTS"), DEFAULTS); +} + void InitConnectivityStateConstants(Handle exports) { NanScope(); Handle channel_state = NanNew(); @@ -183,6 +202,7 @@ void init(Handle exports) { InitStatusConstants(exports); InitCallErrorConstants(exports); InitOpTypeConstants(exports); + InitPropagateConstants(exports); InitConnectivityStateConstants(exports); grpc::node::Call::Init(exports); diff --git a/index.js b/index.js index b26ab35f..93c65ac5 100644 --- a/index.js +++ b/index.js @@ -134,6 +134,11 @@ exports.Server = server.Server; */ exports.status = grpc.status; +/** + * Propagate flag name to number mapping + */ +exports.propagate = grpc.propagate; + /** * Call error name to code number mapping */ diff --git a/src/client.js b/src/client.js index d14713f3..50cbf4a1 100644 --- a/src/client.js +++ b/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/server.js b/src/server.js index 5c62f599..8b86173f 100644 --- a/src/server.js +++ b/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/test/constant_test.js b/test/constant_test.js index 93bf0c8a..fa06ad4e 100644 --- a/test/constant_test.js +++ b/test/constant_test.js @@ -79,6 +79,18 @@ var callErrorNames = [ ]; /** + * List of all propagate flag names + * @const + * @type {Array.} + */ +var propagateFlagNames = [ + 'DEADLINE', + 'CENSUS_STATS_CONTEXT', + 'CENSUS_TRACING_CONTEXT', + 'CANCELLATION', + 'DEFAULTS' +]; +/* * List of all connectivity state names * @const * @type {Array.} @@ -104,6 +116,12 @@ 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]); + } + }); it('should have all of the connectivity states', function() { for (var i = 0; i < connectivityStateNames.length; i++) { assert(grpc.connectivityState.hasOwnProperty(connectivityStateNames[i]), diff --git a/test/surface_test.js b/test/surface_test.js index 098905e7..52515cc8 100644 --- a/test/surface_test.js +++ b/test/surface_test.js @@ -67,6 +67,7 @@ function multiDone(done, count) { } }; } + var server_insecure_creds = grpc.ServerCredentials.createInsecure(); describe('File loader', function() { @@ -344,12 +345,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) { @@ -411,7 +414,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(); }); @@ -664,6 +667,166 @@ 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() { + /* jshint bitwise:false */ + var deadline_flags = (grpc.propagate.DEFAULTS & + ~grpc.propagate.CANCELLATION); + 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.DEADLINE_EXCEEDED); + } finally { + callback(err, value); + done(); + } + }, null, {parent: parent, propagate_flags: deadline_flags}); + }; + 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); + proxy_client.clientStream(function(err, value) { + done(); + }, null, {deadline: deadline}); + }); + it('With a bidi stream call', function(done) { + done = multiDone(done, 2); + proxy_impl.bidiStream = function(parent) { + var child = client.bidiStream( + null, {parent: parent, propagate_flags: deadline_flags}); + 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;