diff --git a/README.md b/README.md index 78781dab..7d3d8c7f 100644 --- a/README.md +++ b/README.md @@ -85,7 +85,7 @@ An object with factory methods for creating credential objects for clients. ServerCredentials ``` -An object with factory methods fro creating credential objects for servers. +An object with factory methods for creating credential objects for servers. [homebrew]:http://brew.sh [linuxbrew]:https://github.com/Homebrew/linuxbrew#installation diff --git a/binding.gyp b/binding.gyp index 6ba23338..734dc841 100644 --- a/binding.gyp +++ b/binding.gyp @@ -11,7 +11,8 @@ '-pedantic', '-g', '-zdefs', - '-Werror' + '-Werror', + '-Wno-error=deprecated-declarations' ], 'ldflags': [ '-g' diff --git a/examples/math_server.js b/examples/math_server.js index b1f8a632..31892c65 100644 --- a/examples/math_server.js +++ b/examples/math_server.js @@ -115,7 +115,7 @@ server.addProtoService(math.Math.service, { }); if (require.main === module) { - server.bind('0.0.0.0:50051'); + server.bind('0.0.0.0:50051', grpc.ServerCredentials.createInsecure()); server.start(); } diff --git a/examples/perf_test.js b/examples/perf_test.js index da919ece..ba8fbf88 100644 --- a/examples/perf_test.js +++ b/examples/perf_test.js @@ -40,8 +40,9 @@ var interop_server = require('../interop/interop_server.js'); function runTest(iterations, callback) { var testServer = interop_server.getServer(0, false); - testServer.server.listen(); - var client = new testProto.TestService('localhost:' + testServer.port); + testServer.server.start(); + var client = new testProto.TestService('localhost:' + testServer.port, + grpc.Credentials.createInsecure()); function runIterations(finish) { var start = process.hrtime(); @@ -62,7 +63,7 @@ function runTest(iterations, callback) { var timeDiff = process.hrtime(startTime); intervals[i] = timeDiff[0] * 1000000 + timeDiff[1] / 1000; next(i+1); - }, {}, deadline); + }, {}, {deadline: deadline}); } } next(0); diff --git a/examples/qps_test.js b/examples/qps_test.js index 00293b46..ec968b85 100644 --- a/examples/qps_test.js +++ b/examples/qps_test.js @@ -60,8 +60,9 @@ var interop_server = require('../interop/interop_server.js'); */ function runTest(concurrent_calls, seconds, callback) { var testServer = interop_server.getServer(0, false); - testServer.server.listen(); - var client = new testProto.TestService('localhost:' + testServer.port); + testServer.server.start(); + var client = new testProto.TestService('localhost:' + testServer.port, + grpc.Credentials.createInsecure()); var warmup_num = 100; diff --git a/examples/route_guide_client.js b/examples/route_guide_client.js index 8cd532fe..647f3ffa 100644 --- a/examples/route_guide_client.js +++ b/examples/route_guide_client.js @@ -40,7 +40,8 @@ var path = require('path'); var _ = require('lodash'); var grpc = require('..'); var examples = grpc.load(__dirname + '/route_guide.proto').examples; -var client = new examples.RouteGuide('localhost:50051'); +var client = new examples.RouteGuide('localhost:50051', + grpc.Credentials.createInsecure()); var COORD_FACTOR = 1e7; diff --git a/examples/route_guide_server.js b/examples/route_guide_server.js index 70044a32..465b32f5 100644 --- a/examples/route_guide_server.js +++ b/examples/route_guide_server.js @@ -239,7 +239,7 @@ function getServer() { if (require.main === module) { // If this is run as a script, start a server on an unused port var routeServer = getServer(); - routeServer.bind('0.0.0.0:50051'); + routeServer.bind('0.0.0.0:50051', grpc.ServerCredentials.createInsecure()); var argv = parseArgs(process.argv, { string: 'db_path' }); @@ -248,7 +248,7 @@ if (require.main === module) { throw err; } feature_list = JSON.parse(data); - routeServer.listen(); + routeServer.start(); }); } diff --git a/examples/stock_client.js b/examples/stock_client.js index b37e66df..ab9b050e 100644 --- a/examples/stock_client.js +++ b/examples/stock_client.js @@ -38,7 +38,8 @@ var examples = grpc.load(__dirname + '/stock.proto').examples; * This exports a client constructor for the Stock service. The usage looks like * * var StockClient = require('stock_client.js'); - * var stockClient = new StockClient(server_address); + * var stockClient = new StockClient(server_address, + * grpc.Credentials.createInsecure()); * stockClient.getLastTradePrice({symbol: 'GOOG'}, function(error, response) { * console.log(error || response); * }); diff --git a/examples/stock_server.js b/examples/stock_server.js index f2eb6ad4..12e54795 100644 --- a/examples/stock_server.js +++ b/examples/stock_server.js @@ -80,8 +80,8 @@ stockServer.addProtoService(examples.Stock.service, { }); if (require.main === module) { - stockServer.bind('0.0.0.0:50051'); - stockServer.listen(); + stockServer.bind('0.0.0.0:50051', grpc.ServerCredentials.createInsecure()); + stockServer.start(); } module.exports = stockServer; diff --git a/ext/call.cc b/ext/call.cc index 7d21b8b4..3256f41d 100644 --- a/ext/call.cc +++ b/ext/call.cc @@ -192,7 +192,7 @@ class SendMetadataOp : public Op { } protected: std::string GetTypeString() const { - return "send metadata"; + return "send_metadata"; } }; @@ -223,7 +223,7 @@ class SendMessageOp : public Op { } protected: std::string GetTypeString() const { - return "send message"; + return "send_message"; } }; @@ -239,7 +239,7 @@ class SendClientCloseOp : public Op { } protected: std::string GetTypeString() const { - return "client close"; + return "client_close"; } }; @@ -283,7 +283,7 @@ class SendServerStatusOp : public Op { } protected: std::string GetTypeString() const { - return "send status"; + return "send_status"; } }; @@ -460,6 +460,8 @@ void Call::Init(Handle exports) { NanNew(StartBatch)->GetFunction()); NanSetPrototypeTemplate(tpl, "cancel", NanNew(Cancel)->GetFunction()); + NanSetPrototypeTemplate(tpl, "getPeer", + NanNew(GetPeer)->GetFunction()); NanAssignPersistent(fun_tpl, tpl); Handle ctr = tpl->GetFunction(); ctr->Set(NanNew("WRITE_BUFFER_HINT"), @@ -507,6 +509,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) { @@ -515,9 +533,21 @@ NAN_METHOD(Call::New) { NanUtf8String method(args[1]); double deadline = args[2]->NumberValue(); grpc_channel *wrapped_channel = channel->GetWrappedChannel(); - grpc_call *wrapped_call = grpc_channel_create_call( - wrapped_channel, CompletionQueueAsyncWorker::GetQueue(), *method, - channel->GetHost(), MillisecondsToTimespec(deadline)); + grpc_call *wrapped_call; + if (args[3]->IsString()) { + NanUtf8String host_override(args[3]); + wrapped_call = grpc_channel_create_call( + 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, parent_call, propagate_flags, + CompletionQueueAsyncWorker::GetQueue(), *method, + NULL, MillisecondsToTimespec(deadline), NULL); + } else { + return NanThrowTypeError("Call's fourth argument must be a string"); + } call = new Call(wrapped_call); args.This()->SetHiddenValue(NanNew("channel_"), channel_object); } @@ -594,7 +624,7 @@ NAN_METHOD(Call::StartBatch) { NanCallback *callback = new NanCallback(callback_func); grpc_call_error error = grpc_call_start_batch( call->wrapped_call, &ops[0], nops, new struct tag( - callback, op_vector.release(), resources)); + callback, op_vector.release(), resources), NULL); if (error != GRPC_CALL_OK) { return NanThrowError("startBatch failed", error); } @@ -608,12 +638,24 @@ NAN_METHOD(Call::Cancel) { return NanThrowTypeError("cancel can only be called on Call objects"); } Call *call = ObjectWrap::Unwrap(args.This()); - grpc_call_error error = grpc_call_cancel(call->wrapped_call); + grpc_call_error error = grpc_call_cancel(call->wrapped_call, NULL); if (error != GRPC_CALL_OK) { return NanThrowError("cancel failed", error); } NanReturnUndefined(); } +NAN_METHOD(Call::GetPeer) { + NanScope(); + if (!HasInstance(args.This())) { + return NanThrowTypeError("getPeer can only be called on Call objects"); + } + Call *call = ObjectWrap::Unwrap(args.This()); + char *peer = grpc_call_get_peer(call->wrapped_call); + Handle peer_value = NanNew(peer); + gpr_free(peer); + NanReturnValue(peer_value); +} + } // namespace node } // namespace grpc diff --git a/ext/call.h b/ext/call.h index 43142c70..6acda761 100644 --- a/ext/call.h +++ b/ext/call.h @@ -120,6 +120,7 @@ class Call : public ::node::ObjectWrap { static NAN_METHOD(New); static NAN_METHOD(StartBatch); static NAN_METHOD(Cancel); + static NAN_METHOD(GetPeer); static NanCallback *constructor; // Used for typechecking instances of this javascript class static v8::Persistent fun_tpl; diff --git a/ext/channel.cc b/ext/channel.cc index d37bf763..a61c8300 100644 --- a/ext/channel.cc +++ b/ext/channel.cc @@ -33,12 +33,17 @@ #include +#include "grpc/support/log.h" + #include #include #include "grpc/grpc.h" #include "grpc/grpc_security.h" +#include "call.h" #include "channel.h" +#include "completion_queue_async_worker.h" #include "credentials.h" +#include "timeval.h" namespace grpc { namespace node { @@ -51,6 +56,7 @@ using v8::Handle; using v8::HandleScope; using v8::Integer; using v8::Local; +using v8::Number; using v8::Object; using v8::Persistent; using v8::String; @@ -59,14 +65,12 @@ using v8::Value; NanCallback *Channel::constructor; Persistent Channel::fun_tpl; -Channel::Channel(grpc_channel *channel, NanUtf8String *host) - : wrapped_channel(channel), host(host) {} +Channel::Channel(grpc_channel *channel) : wrapped_channel(channel) {} Channel::~Channel() { if (wrapped_channel != NULL) { grpc_channel_destroy(wrapped_channel); } - delete host; } void Channel::Init(Handle exports) { @@ -76,6 +80,14 @@ void Channel::Init(Handle exports) { tpl->InstanceTemplate()->SetInternalFieldCount(1); NanSetPrototypeTemplate(tpl, "close", NanNew(Close)->GetFunction()); + NanSetPrototypeTemplate(tpl, "getTarget", + NanNew(GetTarget)->GetFunction()); + NanSetPrototypeTemplate( + tpl, "getConnectivityState", + NanNew(GetConnectivityState)->GetFunction()); + NanSetPrototypeTemplate( + tpl, "watchConnectivityState", + NanNew(WatchConnectivityState)->GetFunction()); NanAssignPersistent(fun_tpl, tpl); Handle ctr = tpl->GetFunction(); constructor = new NanCallback(ctr); @@ -89,38 +101,31 @@ bool Channel::HasInstance(Handle val) { grpc_channel *Channel::GetWrappedChannel() { return this->wrapped_channel; } -char *Channel::GetHost() { return **this->host; } - NAN_METHOD(Channel::New) { NanScope(); if (args.IsConstructCall()) { if (!args[0]->IsString()) { - return NanThrowTypeError("Channel expects a string and an object"); + return NanThrowTypeError( + "Channel expects a string, a credential and an object"); } grpc_channel *wrapped_channel; // Owned by the Channel object - NanUtf8String *host = new NanUtf8String(args[0]); - NanUtf8String *host_override = NULL; - if (args[1]->IsUndefined()) { - wrapped_channel = grpc_channel_create(**host, NULL); - } else if (args[1]->IsObject()) { - grpc_credentials *creds = NULL; - Handle args_hash(args[1]->ToObject()->Clone()); - if (args_hash->HasOwnProperty(NanNew(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG))) { - host_override = new NanUtf8String(args_hash->Get(NanNew(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG))); - } - if (args_hash->HasOwnProperty(NanNew("credentials"))) { - Handle creds_value = args_hash->Get(NanNew("credentials")); - if (!Credentials::HasInstance(creds_value)) { - return NanThrowTypeError( - "credentials arg must be a Credentials object"); - } - Credentials *creds_object = - ObjectWrap::Unwrap(creds_value->ToObject()); - creds = creds_object->GetWrappedCredentials(); - args_hash->Delete(NanNew("credentials")); - } + NanUtf8String host(args[0]); + grpc_credentials *creds; + if (!Credentials::HasInstance(args[1])) { + return NanThrowTypeError( + "Channel's second argument must be a credential"); + } + Credentials *creds_object = ObjectWrap::Unwrap( + args[1]->ToObject()); + creds = creds_object->GetWrappedCredentials(); + grpc_channel_args *channel_args_ptr; + if (args[2]->IsUndefined()) { + channel_args_ptr = NULL; + wrapped_channel = grpc_insecure_channel_create(*host, NULL, NULL); + } else if (args[2]->IsObject()) { + Handle args_hash(args[2]->ToObject()->Clone()); Handle keys(args_hash->GetOwnPropertyNames()); grpc_channel_args channel_args; channel_args.num_args = keys->Length(); @@ -147,27 +152,26 @@ NAN_METHOD(Channel::New) { return NanThrowTypeError("Arg values must be strings"); } } - if (creds == NULL) { - wrapped_channel = grpc_channel_create(**host, &channel_args); - } else { - wrapped_channel = - grpc_secure_channel_create(creds, **host, &channel_args); - } - free(channel_args.args); + channel_args_ptr = &channel_args; } else { return NanThrowTypeError("Channel expects a string and an object"); } - Channel *channel; - if (host_override == NULL) { - channel = new Channel(wrapped_channel, host); + if (creds == NULL) { + wrapped_channel = grpc_insecure_channel_create(*host, channel_args_ptr, + NULL); } else { - channel = new Channel(wrapped_channel, host_override); + wrapped_channel = + grpc_secure_channel_create(creds, *host, channel_args_ptr); } + if (channel_args_ptr != NULL) { + free(channel_args_ptr->args); + } + Channel *channel = new Channel(wrapped_channel); channel->Wrap(args.This()); NanReturnValue(args.This()); } else { - const int argc = 2; - Local argv[argc] = {args[0], args[1]}; + const int argc = 3; + Local argv[argc] = {args[0], args[1], args[2]}; NanReturnValue(constructor->GetFunction()->NewInstance(argc, argv)); } } @@ -185,5 +189,61 @@ NAN_METHOD(Channel::Close) { NanReturnUndefined(); } +NAN_METHOD(Channel::GetTarget) { + NanScope(); + if (!HasInstance(args.This())) { + return NanThrowTypeError("getTarget can only be called on Channel objects"); + } + Channel *channel = ObjectWrap::Unwrap(args.This()); + NanReturnValue(NanNew(grpc_channel_get_target(channel->wrapped_channel))); +} + +NAN_METHOD(Channel::GetConnectivityState) { + NanScope(); + if (!HasInstance(args.This())) { + return NanThrowTypeError( + "getConnectivityState can only be called on Channel objects"); + } + Channel *channel = ObjectWrap::Unwrap(args.This()); + int try_to_connect = (int)args[0]->Equals(NanTrue()); + NanReturnValue(grpc_channel_check_connectivity_state(channel->wrapped_channel, + try_to_connect)); +} + +NAN_METHOD(Channel::WatchConnectivityState) { + NanScope(); + if (!HasInstance(args.This())) { + return NanThrowTypeError( + "watchConnectivityState can only be called on Channel objects"); + } + if (!args[0]->IsUint32()) { + return NanThrowTypeError( + "watchConnectivityState's first argument must be a channel state"); + } + if (!(args[1]->IsNumber() || args[1]->IsDate())) { + return NanThrowTypeError( + "watchConnectivityState's second argument must be a date or a number"); + } + if (!args[2]->IsFunction()) { + return NanThrowTypeError( + "watchConnectivityState's third argument must be a callback"); + } + grpc_connectivity_state last_state = + static_cast(args[0]->Uint32Value()); + double deadline = args[1]->NumberValue(); + Handle callback_func = args[2].As(); + NanCallback *callback = new NanCallback(callback_func); + Channel *channel = ObjectWrap::Unwrap(args.This()); + unique_ptr ops(new OpVec()); + grpc_channel_watch_connectivity_state( + channel->wrapped_channel, last_state, MillisecondsToTimespec(deadline), + CompletionQueueAsyncWorker::GetQueue(), + new struct tag(callback, + ops.release(), + shared_ptr(nullptr))); + CompletionQueueAsyncWorker::Next(); + NanReturnUndefined(); +} + } // namespace node } // namespace grpc diff --git a/ext/channel.h b/ext/channel.h index b3aa0f70..458f71d0 100644 --- a/ext/channel.h +++ b/ext/channel.h @@ -53,11 +53,8 @@ class Channel : public ::node::ObjectWrap { /* Returns the grpc_channel struct that this object wraps */ grpc_channel *GetWrappedChannel(); - /* Return the hostname that this channel connects to */ - char *GetHost(); - private: - explicit Channel(grpc_channel *channel, NanUtf8String *host); + explicit Channel(grpc_channel *channel); ~Channel(); // Prevent copying @@ -66,11 +63,13 @@ class Channel : public ::node::ObjectWrap { static NAN_METHOD(New); static NAN_METHOD(Close); + static NAN_METHOD(GetTarget); + static NAN_METHOD(GetConnectivityState); + static NAN_METHOD(WatchConnectivityState); static NanCallback *constructor; static v8::Persistent fun_tpl; grpc_channel *wrapped_channel; - NanUtf8String *host; }; } // namespace node diff --git a/ext/completion_queue_async_worker.cc b/ext/completion_queue_async_worker.cc index 1215c97e..bf2cd946 100644 --- a/ext/completion_queue_async_worker.cc +++ b/ext/completion_queue_async_worker.cc @@ -63,9 +63,9 @@ CompletionQueueAsyncWorker::~CompletionQueueAsyncWorker() {} void CompletionQueueAsyncWorker::Execute() { result = - grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME)); + grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); if (!result.success) { - SetErrorMessage("The batch encountered an error"); + SetErrorMessage("The async function encountered an error"); } } @@ -85,7 +85,7 @@ void CompletionQueueAsyncWorker::Init(Handle exports) { NanScope(); current_threads = 0; waiting_next_calls = 0; - queue = grpc_completion_queue_create(); + queue = grpc_completion_queue_create(NULL); } void CompletionQueueAsyncWorker::HandleOKCallback() { diff --git a/ext/credentials.cc b/ext/credentials.cc index 34872017..21d61f1a 100644 --- a/ext/credentials.cc +++ b/ext/credentials.cc @@ -79,10 +79,10 @@ void Credentials::Init(Handle exports) { NanNew(CreateComposite)->GetFunction()); ctr->Set(NanNew("createGce"), NanNew(CreateGce)->GetFunction()); - ctr->Set(NanNew("createFake"), - NanNew(CreateFake)->GetFunction()); ctr->Set(NanNew("createIam"), NanNew(CreateIam)->GetFunction()); + ctr->Set(NanNew("createInsecure"), + NanNew(CreateInsecure)->GetFunction()); constructor = new NanCallback(ctr); exports->Set(NanNew("Credentials"), ctr); } @@ -94,9 +94,6 @@ bool Credentials::HasInstance(Handle val) { Handle Credentials::WrapStruct(grpc_credentials *credentials) { NanEscapableScope(); - if (credentials == NULL) { - return NanEscapeScope(NanNull()); - } const int argc = 1; Handle argv[argc] = { NanNew(reinterpret_cast(credentials))}; @@ -130,7 +127,11 @@ NAN_METHOD(Credentials::New) { NAN_METHOD(Credentials::CreateDefault) { NanScope(); - NanReturnValue(WrapStruct(grpc_google_default_credentials_create())); + grpc_credentials *creds = grpc_google_default_credentials_create(); + if (creds == NULL) { + NanReturnNull(); + } + NanReturnValue(WrapStruct(creds)); } NAN_METHOD(Credentials::CreateSsl) { @@ -154,9 +155,12 @@ NAN_METHOD(Credentials::CreateSsl) { return NanThrowTypeError( "createSSl's third argument must be a Buffer if provided"); } - - NanReturnValue(WrapStruct(grpc_ssl_credentials_create( - root_certs, key_cert_pair.private_key == NULL ? NULL : &key_cert_pair))); + grpc_credentials *creds = grpc_ssl_credentials_create( + root_certs, key_cert_pair.private_key == NULL ? NULL : &key_cert_pair); + if (creds == NULL) { + NanReturnNull(); + } + NanReturnValue(WrapStruct(creds)); } NAN_METHOD(Credentials::CreateComposite) { @@ -171,18 +175,21 @@ NAN_METHOD(Credentials::CreateComposite) { } Credentials *creds1 = ObjectWrap::Unwrap(args[0]->ToObject()); Credentials *creds2 = ObjectWrap::Unwrap(args[1]->ToObject()); - NanReturnValue(WrapStruct(grpc_composite_credentials_create( - creds1->wrapped_credentials, creds2->wrapped_credentials))); + grpc_credentials *creds = grpc_composite_credentials_create( + creds1->wrapped_credentials, creds2->wrapped_credentials); + if (creds == NULL) { + NanReturnNull(); + } + NanReturnValue(WrapStruct(creds)); } NAN_METHOD(Credentials::CreateGce) { NanScope(); - NanReturnValue(WrapStruct(grpc_compute_engine_credentials_create())); -} - -NAN_METHOD(Credentials::CreateFake) { - NanScope(); - NanReturnValue(WrapStruct(grpc_fake_transport_security_credentials_create())); + grpc_credentials *creds = grpc_compute_engine_credentials_create(); + if (creds == NULL) { + NanReturnNull(); + } + NanReturnValue(WrapStruct(creds)); } NAN_METHOD(Credentials::CreateIam) { @@ -195,8 +202,17 @@ NAN_METHOD(Credentials::CreateIam) { } NanUtf8String auth_token(args[0]); NanUtf8String auth_selector(args[1]); - NanReturnValue( - WrapStruct(grpc_iam_credentials_create(*auth_token, *auth_selector))); + grpc_credentials *creds = grpc_iam_credentials_create(*auth_token, + *auth_selector); + if (creds == NULL) { + NanReturnNull(); + } + NanReturnValue(WrapStruct(creds)); +} + +NAN_METHOD(Credentials::CreateInsecure) { + NanScope(); + NanReturnValue(WrapStruct(NULL)); } } // namespace node diff --git a/ext/credentials.h b/ext/credentials.h index 794736fe..62957e61 100644 --- a/ext/credentials.h +++ b/ext/credentials.h @@ -68,6 +68,7 @@ class Credentials : public ::node::ObjectWrap { static NAN_METHOD(CreateGce); static NAN_METHOD(CreateFake); static NAN_METHOD(CreateIam); + static NAN_METHOD(CreateInsecure); static NanCallback *constructor; // Used for typechecking instances of this javascript class static v8::Persistent fun_tpl; diff --git a/ext/node_grpc.cc b/ext/node_grpc.cc index 4e31cbaa..d93dafda 100644 --- a/ext/node_grpc.cc +++ b/ext/node_grpc.cc @@ -159,12 +159,51 @@ 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(); + exports->Set(NanNew("connectivityState"), channel_state); + Handle IDLE(NanNew(GRPC_CHANNEL_IDLE)); + channel_state->Set(NanNew("IDLE"), IDLE); + Handle CONNECTING(NanNew(GRPC_CHANNEL_CONNECTING)); + channel_state->Set(NanNew("CONNECTING"), CONNECTING); + Handle READY(NanNew(GRPC_CHANNEL_READY)); + channel_state->Set(NanNew("READY"), READY); + Handle TRANSIENT_FAILURE( + NanNew(GRPC_CHANNEL_TRANSIENT_FAILURE)); + channel_state->Set(NanNew("TRANSIENT_FAILURE"), TRANSIENT_FAILURE); + Handle FATAL_FAILURE( + NanNew(GRPC_CHANNEL_FATAL_FAILURE)); + channel_state->Set(NanNew("FATAL_FAILURE"), FATAL_FAILURE); +} + void init(Handle exports) { NanScope(); grpc_init(); InitStatusConstants(exports); InitCallErrorConstants(exports); InitOpTypeConstants(exports); + InitPropagateConstants(exports); + InitConnectivityStateConstants(exports); grpc::node::Call::Init(exports); grpc::node::Channel::Init(exports); diff --git a/ext/server.cc b/ext/server.cc index 34cde9ff..8e396448 100644 --- a/ext/server.cc +++ b/ext/server.cc @@ -108,13 +108,13 @@ class NewCallOp : public Op { protected: std::string GetTypeString() const { - return "new call"; + return "new_call"; } }; Server::Server(grpc_server *server) : wrapped_server(server) { - shutdown_queue = grpc_completion_queue_create(); - grpc_server_register_completion_queue(server, shutdown_queue); + shutdown_queue = grpc_completion_queue_create(NULL); + grpc_server_register_completion_queue(server, shutdown_queue, NULL); } Server::~Server() { @@ -136,10 +136,6 @@ void Server::Init(Handle exports) { tpl, "addHttp2Port", NanNew(AddHttp2Port)->GetFunction()); - NanSetPrototypeTemplate( - tpl, "addSecureHttp2Port", - NanNew(AddSecureHttp2Port)->GetFunction()); - NanSetPrototypeTemplate(tpl, "start", NanNew(Start)->GetFunction()); @@ -162,7 +158,7 @@ void Server::ShutdownServer() { this->shutdown_queue, NULL); grpc_completion_queue_pluck(this->shutdown_queue, NULL, - gpr_inf_future(GPR_CLOCK_REALTIME)); + gpr_inf_future(GPR_CLOCK_REALTIME), NULL); this->wrapped_server = NULL; } } @@ -180,7 +176,7 @@ NAN_METHOD(Server::New) { grpc_server *wrapped_server; grpc_completion_queue *queue = CompletionQueueAsyncWorker::GetQueue(); if (args[0]->IsUndefined()) { - wrapped_server = grpc_server_create(NULL); + wrapped_server = grpc_server_create(NULL, NULL); } else if (args[0]->IsObject()) { Handle args_hash(args[0]->ToObject()); Handle keys(args_hash->GetOwnPropertyNames()); @@ -209,12 +205,12 @@ NAN_METHOD(Server::New) { return NanThrowTypeError("Arg values must be strings"); } } - wrapped_server = grpc_server_create(&channel_args); + wrapped_server = grpc_server_create(&channel_args, NULL); free(channel_args.args); } else { return NanThrowTypeError("Server expects an object"); } - grpc_server_register_completion_queue(wrapped_server, queue); + grpc_server_register_completion_queue(wrapped_server, queue, NULL); Server *server = new Server(wrapped_server); server->Wrap(args.This()); NanReturnValue(args.This()); @@ -246,45 +242,37 @@ NAN_METHOD(Server::RequestCall) { } NAN_METHOD(Server::AddHttp2Port) { - NanScope(); - if (!HasInstance(args.This())) { - return NanThrowTypeError("addHttp2Port can only be called on a Server"); - } - if (!args[0]->IsString()) { - return NanThrowTypeError("addHttp2Port's argument must be a String"); - } - Server *server = ObjectWrap::Unwrap(args.This()); - if (server->wrapped_server == NULL) { - return NanThrowError("addHttp2Port cannot be called on a shut down Server"); - } - NanReturnValue(NanNew(grpc_server_add_http2_port( - server->wrapped_server, *NanUtf8String(args[0])))); -} - -NAN_METHOD(Server::AddSecureHttp2Port) { NanScope(); if (!HasInstance(args.This())) { return NanThrowTypeError( - "addSecureHttp2Port can only be called on a Server"); + "addHttp2Port can only be called on a Server"); } if (!args[0]->IsString()) { return NanThrowTypeError( - "addSecureHttp2Port's first argument must be a String"); + "addHttp2Port's first argument must be a String"); } if (!ServerCredentials::HasInstance(args[1])) { return NanThrowTypeError( - "addSecureHttp2Port's second argument must be ServerCredentials"); + "addHttp2Port's second argument must be ServerCredentials"); } Server *server = ObjectWrap::Unwrap(args.This()); if (server->wrapped_server == NULL) { return NanThrowError( - "addSecureHttp2Port cannot be called on a shut down Server"); + "addHttp2Port cannot be called on a shut down Server"); } - ServerCredentials *creds = ObjectWrap::Unwrap( + ServerCredentials *creds_object = ObjectWrap::Unwrap( args[1]->ToObject()); - NanReturnValue(NanNew(grpc_server_add_secure_http2_port( - server->wrapped_server, *NanUtf8String(args[0]), - creds->GetWrappedServerCredentials()))); + grpc_server_credentials *creds = creds_object->GetWrappedServerCredentials(); + int port; + if (creds == NULL) { + port = grpc_server_add_insecure_http2_port(server->wrapped_server, + *NanUtf8String(args[0])); + } else { + port = grpc_server_add_secure_http2_port(server->wrapped_server, + *NanUtf8String(args[0]), + creds); + } + NanReturnValue(NanNew(port)); } NAN_METHOD(Server::Start) { diff --git a/ext/server.h b/ext/server.h index 5b4b18a0..faab7e34 100644 --- a/ext/server.h +++ b/ext/server.h @@ -66,7 +66,6 @@ class Server : public ::node::ObjectWrap { static NAN_METHOD(New); static NAN_METHOD(RequestCall); static NAN_METHOD(AddHttp2Port); - static NAN_METHOD(AddSecureHttp2Port); static NAN_METHOD(Start); static NAN_METHOD(Shutdown); static NanCallback *constructor; diff --git a/ext/server_credentials.cc b/ext/server_credentials.cc index d2b63cdc..1b8e7b43 100644 --- a/ext/server_credentials.cc +++ b/ext/server_credentials.cc @@ -73,8 +73,8 @@ void ServerCredentials::Init(Handle exports) { Handle ctr = tpl->GetFunction(); ctr->Set(NanNew("createSsl"), NanNew(CreateSsl)->GetFunction()); - ctr->Set(NanNew("createFake"), - NanNew(CreateFake)->GetFunction()); + ctr->Set(NanNew("createInsecure"), + NanNew(CreateInsecure)->GetFunction()); constructor = new NanCallback(ctr); exports->Set(NanNew("ServerCredentials"), ctr); } @@ -87,9 +87,6 @@ bool ServerCredentials::HasInstance(Handle val) { Handle ServerCredentials::WrapStruct( grpc_server_credentials *credentials) { NanEscapableScope(); - if (credentials == NULL) { - return NanEscapeScope(NanNull()); - } const int argc = 1; Handle argv[argc] = { NanNew(reinterpret_cast(credentials))}; @@ -140,14 +137,19 @@ NAN_METHOD(ServerCredentials::CreateSsl) { return NanThrowTypeError("createSsl's third argument must be a Buffer"); } key_cert_pair.cert_chain = ::node::Buffer::Data(args[2]); - NanReturnValue(WrapStruct( - grpc_ssl_server_credentials_create(root_certs, &key_cert_pair, 1))); + // TODO Add a force_client_auth parameter and pass it as the last parameter + // here. + grpc_server_credentials *creds = + grpc_ssl_server_credentials_create(root_certs, &key_cert_pair, 1, 0); + if (creds == NULL) { + NanReturnNull(); + } + NanReturnValue(WrapStruct(creds)); } -NAN_METHOD(ServerCredentials::CreateFake) { +NAN_METHOD(ServerCredentials::CreateInsecure) { NanScope(); - NanReturnValue( - WrapStruct(grpc_fake_transport_security_server_credentials_create())); + NanReturnValue(WrapStruct(NULL)); } } // namespace node diff --git a/ext/server_credentials.h b/ext/server_credentials.h index aaa7ef29..63903f66 100644 --- a/ext/server_credentials.h +++ b/ext/server_credentials.h @@ -63,7 +63,7 @@ class ServerCredentials : public ::node::ObjectWrap { static NAN_METHOD(New); static NAN_METHOD(CreateSsl); - static NAN_METHOD(CreateFake); + static NAN_METHOD(CreateInsecure); static NanCallback *constructor; // Used for typechecking instances of this javascript class static v8::Persistent fun_tpl; diff --git a/ext/timeval.cc b/ext/timeval.cc index 60de4d81..bf68513c 100644 --- a/ext/timeval.cc +++ b/ext/timeval.cc @@ -52,6 +52,7 @@ gpr_timespec MillisecondsToTimespec(double millis) { } double TimespecToMilliseconds(gpr_timespec timespec) { + timespec = gpr_convert_clock_type(timespec, GPR_CLOCK_REALTIME); if (gpr_time_cmp(timespec, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { return std::numeric_limits::infinity(); } else if (gpr_time_cmp(timespec, gpr_inf_past(GPR_CLOCK_REALTIME)) == 0) { diff --git a/index.js b/index.js index d81e7804..93c65ac5 100644 --- a/index.js +++ b/index.js @@ -48,7 +48,7 @@ var grpc = require('bindings')('grpc'); * @param {ProtoBuf.Reflect.Namespace} value The ProtoBuf object to load. * @return {Object} The resulting gRPC object */ -function loadObject(value) { +exports.loadObject = function loadObject(value) { var result = {}; if (value.className === 'Namespace') { _.each(value.children, function(child) { @@ -62,7 +62,9 @@ function loadObject(value) { } else { return value; } -} +}; + +var loadObject = exports.loadObject; /** * Load a gRPC object from a .proto file. @@ -71,7 +73,7 @@ function loadObject(value) { * 'json'. Defaults to 'proto' * @return {Object} The resulting gRPC object */ -function load(filename, format) { +exports.load = function load(filename, format) { if (!format) { format = 'proto'; } @@ -88,7 +90,7 @@ function load(filename, format) { } return loadObject(builder.ns); -} +}; /** * Get a function that a client can use to update metadata with authentication @@ -97,7 +99,7 @@ function load(filename, format) { * @param {Object} credential The credential object to use * @return {function(Object, callback)} Metadata updater function */ -function getGoogleAuthDelegate(credential) { +exports.getGoogleAuthDelegate = function getGoogleAuthDelegate(credential) { /** * Update a metadata object with authentication information. * @param {string} authURI The uri to authenticate to @@ -120,20 +122,10 @@ function getGoogleAuthDelegate(credential) { callback(null, metadata); }); }; -} +}; /** - * See docs for loadObject - */ -exports.loadObject = loadObject; - -/** - * See docs for load - */ -exports.load = load; - -/** - * See docs for Server + * @see module:src/server.Server */ exports.Server = server.Server; @@ -141,6 +133,12 @@ exports.Server = server.Server; * Status name to code number mapping */ exports.status = grpc.status; + +/** + * Propagate flag name to number mapping + */ +exports.propagate = grpc.propagate; + /** * Call error name to code number mapping */ @@ -156,6 +154,7 @@ exports.Credentials = grpc.Credentials; */ exports.ServerCredentials = grpc.ServerCredentials; -exports.getGoogleAuthDelegate = getGoogleAuthDelegate; - +/** + * @see module:src/client.makeClientConstructor + */ exports.makeGenericClientConstructor = client.makeClientConstructor; diff --git a/interop/interop_client.js b/interop/interop_client.js index e810e68e..6d6f9a34 100644 --- a/interop/interop_client.js +++ b/interop/interop_client.js @@ -67,11 +67,8 @@ function zeroBuffer(size) { * primarily for use with mocha */ function emptyUnary(client, done) { - var call = client.emptyCall({}, function(err, resp) { + client.emptyCall({}, function(err, resp) { assert.ifError(err); - }); - call.on('status', function(status) { - assert.strictEqual(status.code, grpc.status.OK); if (done) { done(); } @@ -92,13 +89,10 @@ function largeUnary(client, done) { body: zeroBuffer(271828) } }; - var call = client.unaryCall(arg, function(err, resp) { + client.unaryCall(arg, function(err, resp) { assert.ifError(err); assert.strictEqual(resp.payload.type, 'COMPRESSABLE'); assert.strictEqual(resp.payload.body.length, 314159); - }); - call.on('status', function(status) { - assert.strictEqual(status.code, grpc.status.OK); if (done) { done(); } @@ -115,9 +109,6 @@ function clientStreaming(client, done) { var call = client.streamingInputCall(function(err, resp) { assert.ifError(err); assert.strictEqual(resp.aggregated_payload_size, 74922); - }); - call.on('status', function(status) { - assert.strictEqual(status.code, grpc.status.OK); if (done) { done(); } @@ -268,7 +259,7 @@ function cancelAfterFirstResponse(client, done) { function timeoutOnSleepingServer(client, done) { var deadline = new Date(); deadline.setMilliseconds(deadline.getMilliseconds() + 1); - var call = client.fullDuplexCall(null, deadline); + var call = client.fullDuplexCall(null, {deadline: deadline}); call.write({ payload: {body: zeroBuffer(27182)} }); @@ -302,15 +293,14 @@ function authTest(expected_user, scope, client, done) { fill_username: true, fill_oauth_scope: true }; - var call = client.unaryCall(arg, function(err, resp) { + client.unaryCall(arg, function(err, resp) { assert.ifError(err); assert.strictEqual(resp.payload.type, 'COMPRESSABLE'); assert.strictEqual(resp.payload.body.length, 314159); assert.strictEqual(resp.username, expected_user); - assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE); - }); - call.on('status', function(status) { - assert.strictEqual(status.code, grpc.status.OK); + if (scope) { + assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE); + } if (done) { done(); } @@ -340,17 +330,14 @@ function oauth2Test(expected_user, scope, per_rpc, client, done) { }; var makeTestCall = function(error, client_metadata) { assert.ifError(error); - var call = client.unaryCall(arg, function(err, resp) { + client.unaryCall(arg, function(err, resp) { assert.ifError(err); assert.strictEqual(resp.username, expected_user); assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE); - }); - call.on('status', function(status) { - assert.strictEqual(status.code, grpc.status.OK); if (done) { done(); } - }); + }, client_metadata); }; if (per_rpc) { updateMetadata('', {}, makeTestCall); @@ -358,7 +345,6 @@ function oauth2Test(expected_user, scope, per_rpc, client, done) { client.updateMetadata = updateMetadata; makeTestCall(null, {}); } - }); }); } @@ -397,6 +383,7 @@ var test_cases = { function runTest(address, host_override, test_case, tls, test_ca, done) { // TODO(mlumish): enable TLS functionality var options = {}; + var creds; if (tls) { var ca_path; if (test_ca) { @@ -405,13 +392,15 @@ function runTest(address, host_override, test_case, tls, test_ca, done) { ca_path = process.env.SSL_CERT_FILE; } var ca_data = fs.readFileSync(ca_path); - var creds = grpc.Credentials.createSsl(ca_data); - options.credentials = creds; + creds = grpc.Credentials.createSsl(ca_data); if (host_override) { options['grpc.ssl_target_name_override'] = host_override; + options['grpc.default_authority'] = host_override; } + } else { + creds = grpc.Credentials.createInsecure(); } - var client = new testProto.TestService(address, options); + var client = new testProto.TestService(address, creds, options); test_cases[test_case](client, done); } diff --git a/interop/interop_server.js b/interop/interop_server.js index 505c6bb5..1242a0f9 100644 --- a/interop/interop_server.js +++ b/interop/interop_server.js @@ -161,7 +161,7 @@ function handleHalfDuplex(call) { function getServer(port, tls) { // TODO(mlumish): enable TLS functionality var options = {}; - var server_creds = null; + var server_creds; if (tls) { var key_path = path.join(__dirname, '../test/data/server1.key'); var pem_path = path.join(__dirname, '../test/data/server1.pem'); @@ -171,6 +171,8 @@ function getServer(port, tls) { server_creds = grpc.ServerCredentials.createSsl(null, key_data, pem_data); + } else { + server_creds = grpc.ServerCredentials.createInsecure(); } var server = new grpc.Server(options); server.addProtoService(testProto.TestService.service, { @@ -192,7 +194,7 @@ if (require.main === module) { }); var server_obj = getServer(argv.port, argv.use_tls === 'true'); console.log('Server attaching to port ' + argv.port); - server_obj.server.listen(); + server_obj.server.start(); } /** diff --git a/jsdoc_conf.json b/jsdoc_conf.json new file mode 100644 index 00000000..876a8e19 --- /dev/null +++ b/jsdoc_conf.json @@ -0,0 +1,22 @@ +{ + "tags": { + "allowUnknownTags": true + }, + "source": { + "include": [ "index.js", "src" ], + "includePattern": ".+\\.js(doc)?$", + "excludePattern": "(^|\\/|\\\\)_" + }, + "opts": { + "package": "package.json", + "readme": "README.md" + }, + "plugins": [], + "templates": { + "cleverLinks": false, + "monospaceLinks": false, + "default": { + "outputSourceFiles": true + } + } +} diff --git a/package.json b/package.json index 1caf1587..756d41b0 100644 --- a/package.json +++ b/package.json @@ -21,7 +21,8 @@ }, "scripts": { "lint": "node ./node_modules/jshint/bin/jshint src test examples interop index.js", - "test": "node ./node_modules/mocha/bin/mocha && npm run-script lint" + "test": "node ./node_modules/mocha/bin/mocha && npm run-script lint", + "gen_docs": "./node_modules/.bin/jsdoc -c jsdoc_conf.json" }, "dependencies": { "bindings": "^1.2.0", @@ -32,6 +33,7 @@ "devDependencies": { "async": "^0.9.0", "google-auth-library": "^0.9.2", + "jsdoc": "^3.3.2", "jshint": "^2.5.0", "minimist": "^1.1.0", "mocha": "~1.21.0", diff --git a/src/client.js b/src/client.js index de9efd7e..3c3642ad 100644 --- a/src/client.js +++ b/src/client.js @@ -31,6 +31,11 @@ * */ +/** + * Server module + * @module + */ + 'use strict'; var _ = require('lodash'); @@ -47,6 +52,7 @@ var Readable = stream.Readable; var Writable = stream.Writable; var Duplex = stream.Duplex; var util = require('util'); +var version = require('../package.json').version; util.inherits(ClientWritableStream, Writable); @@ -71,6 +77,7 @@ function ClientWritableStream(call, serialize) { /** * Attempt to write the given chunk. Calls the callback when done. This is an * implementation of a method needed for implementing stream.Writable. + * @access private * @param {Buffer} chunk The chunk to write * @param {string} encoding Used to pass write flags * @param {function(Error=)} callback Called when the write is complete @@ -111,6 +118,7 @@ function ClientReadableStream(call, deserialize) { /** * Read the next object from the stream. + * @access private * @param {*} size Ignored because we use objectMode=true */ function _read(size) { @@ -188,6 +196,43 @@ ClientReadableStream.prototype.cancel = cancel; ClientWritableStream.prototype.cancel = cancel; ClientDuplexStream.prototype.cancel = cancel; +/** + * Get the endpoint this call/stream is connected to. + * @return {string} The URI of the endpoint + */ +function getPeer() { + /* jshint validthis: true */ + return this.call.getPeer(); +} + +ClientReadableStream.prototype.getPeer = getPeer; +ClientWritableStream.prototype.getPeer = getPeer; +ClientDuplexStream.prototype.getPeer = getPeer; + +/** + * Get a call object built with the provided options. Keys for options are + * 'deadline', which takes a date or number, and 'host', which takes a string + * and overrides the hostname to connect to. + * @param {Object} options Options map. + */ +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, + parent, propagate_flags); +} + /** * Get a function that can make unary requests to the specified method. * @param {string} method The name of the method to request @@ -207,25 +252,22 @@ function makeUnaryRequestFunction(method, serialize, deserialize) { * response is received * @param {array=} metadata Array of metadata key/value pairs to add to the * call - * @param {(number|Date)=} deadline The deadline for processing this request. - * Defaults to infinite future - * @param {number=} flags Flags for modifying how the message is sent. - * Defaults to 0. + * @param {Object=} options Options map * @return {EventEmitter} An event emitter for stream related events */ - function makeUnaryRequest(argument, callback, metadata, deadline, flags) { + function makeUnaryRequest(argument, callback, metadata, options) { /* jshint validthis: true */ - if (deadline === undefined) { - deadline = Infinity; - } var emitter = new EventEmitter(); - var call = new grpc.Call(this.channel, method, deadline); + var call = getCall(this.channel, method, options); if (metadata === null || metadata === undefined) { metadata = {}; } emitter.cancel = function cancel() { call.cancel(); }; + emitter.getPeer = function getPeer() { + return call.getPeer(); + }; this.updateMetadata(this.auth_uri, metadata, function(error, metadata) { if (error) { call.cancel(); @@ -234,7 +276,7 @@ function makeUnaryRequestFunction(method, serialize, deserialize) { } var client_batch = {}; var message = serialize(argument); - message.grpcWriteFlags = flags; + message.grpcWriteFlags = options.flags; client_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata; client_batch[grpc.opType.SEND_MESSAGE] = message; client_batch[grpc.opType.SEND_CLOSE_FROM_CLIENT] = true; @@ -282,16 +324,12 @@ function makeClientStreamRequestFunction(method, serialize, deserialize) { * response is received * @param {array=} metadata Array of metadata key/value pairs to add to the * call - * @param {(number|Date)=} deadline The deadline for processing this request. - * Defaults to infinite future + * @param {Object=} options Options map * @return {EventEmitter} An event emitter for stream related events */ - function makeClientStreamRequest(callback, metadata, deadline) { + function makeClientStreamRequest(callback, metadata, options) { /* jshint validthis: true */ - if (deadline === undefined) { - deadline = Infinity; - } - var call = new grpc.Call(this.channel, method, deadline); + var call = getCall(this.channel, method, options); if (metadata === null || metadata === undefined) { metadata = {}; } @@ -356,18 +394,12 @@ function makeServerStreamRequestFunction(method, serialize, deserialize) { * serialize * @param {array=} metadata Array of metadata key/value pairs to add to the * call - * @param {(number|Date)=} deadline The deadline for processing this request. - * Defaults to infinite future - * @param {number=} flags Flags for modifying how the message is sent. - * Defaults to 0. + * @param {Object} options Options map * @return {EventEmitter} An event emitter for stream related events */ - function makeServerStreamRequest(argument, metadata, deadline, flags) { + function makeServerStreamRequest(argument, metadata, options) { /* jshint validthis: true */ - if (deadline === undefined) { - deadline = Infinity; - } - var call = new grpc.Call(this.channel, method, deadline); + var call = getCall(this.channel, method, options); if (metadata === null || metadata === undefined) { metadata = {}; } @@ -380,7 +412,7 @@ function makeServerStreamRequestFunction(method, serialize, deserialize) { } var start_batch = {}; var message = serialize(argument); - message.grpcWriteFlags = flags; + message.grpcWriteFlags = options.flags; start_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata; start_batch[grpc.opType.RECV_INITIAL_METADATA] = true; start_batch[grpc.opType.SEND_MESSAGE] = message; @@ -432,16 +464,12 @@ function makeBidiStreamRequestFunction(method, serialize, deserialize) { * @this {SurfaceClient} Client object. Must have a channel member. * @param {array=} metadata Array of metadata key/value pairs to add to the * call - * @param {(number|Date)=} deadline The deadline for processing this request. - * Defaults to infinite future + * @param {Options} options Options map * @return {EventEmitter} An event emitter for stream related events */ - function makeBidiStreamRequest(metadata, deadline) { + function makeBidiStreamRequest(metadata, options) { /* jshint validthis: true */ - if (deadline === undefined) { - deadline = Infinity; - } - var call = new grpc.Call(this.channel, method, deadline); + var call = getCall(this.channel, method, options); if (metadata === null || metadata === undefined) { metadata = {}; } @@ -509,31 +537,67 @@ var requester_makers = { * requestSerialize: function to serialize request objects * responseDeserialize: function to deserialize response objects * @param {Object} methods An object mapping method names to method attributes - * @param {string} serviceName The name of the service + * @param {string} serviceName The fully qualified name of the service * @return {function(string, Object)} New client constructor */ -function makeClientConstructor(methods, serviceName) { +exports.makeClientConstructor = function(methods, serviceName) { /** * Create a client with the given methods * @constructor * @param {string} address The address of the server to connect to + * @param {grpc.Credentials} credentials Credentials to use to connect + * to the server * @param {Object} options Options to pass to the underlying channel * @param {function(string, Object, function)=} updateMetadata function to * update the metadata for each request */ - function Client(address, options, updateMetadata) { + function Client(address, credentials, options, updateMetadata) { if (!updateMetadata) { updateMetadata = function(uri, metadata, callback) { callback(null, metadata); }; } - - this.server_address = address.replace(/\/$/, ''); - this.channel = new grpc.Channel(address, options); - this.auth_uri = this.server_address + '/' + serviceName; + if (!options) { + options = {}; + } + options['grpc.primary_user_agent'] = 'grpc-node/' + version; + this.channel = new grpc.Channel(address, credentials, options); + // Remove the optional DNS scheme, trailing port, and trailing backslash + address = address.replace(/^(dns:\/{3})?([^:\/]+)(:\d+)?\/?$/, '$2'); + this.server_address = address; + this.auth_uri = 'https://' + this.server_address + '/' + serviceName; this.updateMetadata = updateMetadata; } + /** + * Wait for the client to be ready. The callback will be called when the + * client has successfully connected to the server, and it will be called + * with an error if the attempt to connect to the server has unrecoverablly + * failed or if the deadline expires. This function will make the channel + * start connecting if it has not already done so. + * @param {(Date|Number)} deadline When to stop waiting for a connection. Pass + * Infinity to wait forever. + * @param {function(Error)} callback The callback to call when done attempting + * to connect. + */ + Client.prototype.$waitForReady = function(deadline, callback) { + var self = this; + var checkState = function(err) { + if (err) { + callback(new Error('Failed to connect before the deadline')); + } + var new_state = self.channel.getConnectivityState(true); + if (new_state === grpc.connectivityState.READY) { + callback(); + } else if (new_state === grpc.connectivityState.FATAL_FAILURE) { + callback(new Error('Failed to connect to server')); + } else { + self.channel.watchConnectivityState(new_state, deadline, checkState); + } + }; + checkState(); + }; + _.each(methods, function(attrs, name) { var method_type; if (attrs.requestStream) { @@ -558,7 +622,7 @@ function makeClientConstructor(methods, serviceName) { }); return Client; -} +}; /** * Creates a constructor for clients for the given service @@ -566,22 +630,19 @@ function makeClientConstructor(methods, serviceName) { * for * @return {function(string, Object)} New client constructor */ -function makeProtobufClientConstructor(service) { +exports.makeProtobufClientConstructor = function(service) { var method_attrs = common.getProtobufServiceAttrs(service, service.name); - var Client = makeClientConstructor(method_attrs); + var Client = exports.makeClientConstructor( + method_attrs, common.fullyQualifiedName(service)); Client.service = service; - return Client; -} - -exports.makeClientConstructor = makeClientConstructor; - -exports.makeProtobufClientConstructor = makeProtobufClientConstructor; +}; /** - * See docs for client.status + * Map of status code names to status codes */ exports.status = grpc.status; + /** * See docs for client.callError */ diff --git a/src/common.js b/src/common.js index feaa859a..5551ebee 100644 --- a/src/common.js +++ b/src/common.js @@ -31,6 +31,10 @@ * */ +/** + * @module + */ + 'use strict'; var _ = require('lodash'); @@ -40,7 +44,7 @@ var _ = require('lodash'); * @param {function()} cls The constructor of the message type to deserialize * @return {function(Buffer):cls} The deserialization function */ -function deserializeCls(cls) { +exports.deserializeCls = function deserializeCls(cls) { /** * Deserialize a buffer to a message object * @param {Buffer} arg_buf The buffer to deserialize @@ -51,14 +55,16 @@ function deserializeCls(cls) { // and longs as strings (second argument) return cls.decode(arg_buf).toRaw(false, true); }; -} +}; + +var deserializeCls = exports.deserializeCls; /** * Get a function that serializes objects to a buffer by protobuf class. * @param {function()} Cls The constructor of the message type to serialize * @return {function(Cls):Buffer} The serialization function */ -function serializeCls(Cls) { +exports.serializeCls = function serializeCls(Cls) { /** * Serialize an object to a Buffer * @param {Object} arg The object to serialize @@ -67,14 +73,16 @@ function serializeCls(Cls) { return function serialize(arg) { return new Buffer(new Cls(arg).encode().toBuffer()); }; -} +}; + +var serializeCls = exports.serializeCls; /** * Get the fully qualified (dotted) name of a ProtoBuf.Reflect value. * @param {ProtoBuf.Reflect.Namespace} value The value to get the name of * @return {string} The fully qualified name of the value */ -function fullyQualifiedName(value) { +exports.fullyQualifiedName = function fullyQualifiedName(value) { if (value === null || value === undefined) { return ''; } @@ -89,7 +97,9 @@ function fullyQualifiedName(value) { } } return name; -} +}; + +var fullyQualifiedName = exports.fullyQualifiedName; /** * Wrap a function to pass null-like values through without calling it. If no @@ -97,7 +107,7 @@ function fullyQualifiedName(value) { * @param {?function} func The function to wrap * @return {function} The wrapped function */ -function wrapIgnoreNull(func) { +exports.wrapIgnoreNull = function wrapIgnoreNull(func) { if (!func) { return _.identity; } @@ -107,14 +117,14 @@ function wrapIgnoreNull(func) { } return func(arg); }; -} +}; /** * Return a map from method names to method attributes for the service. * @param {ProtoBuf.Reflect.Service} service The service to get attributes for * @return {Object} The attributes map */ -function getProtobufServiceAttrs(service) { +exports.getProtobufServiceAttrs = function getProtobufServiceAttrs(service) { var prefix = '/' + fullyQualifiedName(service) + '/'; return _.object(_.map(service.children, function(method) { return [_.camelCase(method.name), { @@ -127,26 +137,4 @@ function getProtobufServiceAttrs(service) { responseDeserialize: deserializeCls(method.resolvedResponseType.build()) }]; })); -} - -/** - * See docs for deserializeCls - */ -exports.deserializeCls = deserializeCls; - -/** - * See docs for serializeCls - */ -exports.serializeCls = serializeCls; - -/** - * See docs for fullyQualifiedName - */ -exports.fullyQualifiedName = fullyQualifiedName; - -/** - * See docs for wrapIgnoreNull - */ -exports.wrapIgnoreNull = wrapIgnoreNull; - -exports.getProtobufServiceAttrs = getProtobufServiceAttrs; +}; diff --git a/src/server.js b/src/server.js index 776fafb9..3b3bab82 100644 --- a/src/server.js +++ b/src/server.js @@ -31,6 +31,11 @@ * */ +/** + * Server module + * @module + */ + 'use strict'; var _ = require('lodash'); @@ -50,6 +55,7 @@ var EventEmitter = require('events').EventEmitter; /** * Handle an error on a call by sending it as a status + * @access private * @param {grpc.Call} call The call to send the error on * @param {Object} error The error object */ @@ -82,6 +88,7 @@ function handleError(call, error) { /** * Wait for the client to close, then emit a cancelled event if the client * cancelled. + * @access private * @param {grpc.Call} call The call object to wait on * @param {EventEmitter} emitter The event emitter to emit the cancelled event * on @@ -102,6 +109,7 @@ function waitForCancel(call, emitter) { /** * Send a response to a unary or client streaming call. + * @access private * @param {grpc.Call} call The call to respond on * @param {*} value The value to respond with * @param {function(*):Buffer=} serialize Serialization function for the @@ -134,6 +142,7 @@ function sendUnaryResponse(call, value, serialize, metadata, flags) { /** * Initialize a writable stream. This is used for both the writable and duplex * stream constructors. + * @access private * @param {Writable} stream The stream to set up * @param {function(*):Buffer=} Serialization function for responses */ @@ -207,6 +216,7 @@ function setUpWritable(stream, serialize) { /** * Initialize a readable stream. This is used for both the readable and duplex * stream constructors. + * @access private * @param {Readable} stream The stream to initialize * @param {function(Buffer):*=} deserialize Deserialization function for * incoming data. @@ -246,6 +256,7 @@ function ServerWritableStream(call, serialize) { /** * Start writing a chunk of data. This is an implementation of a method required * for implementing stream.Writable. + * @access private * @param {Buffer} chunk The chunk of data to write * @param {string} encoding Used to pass write flags * @param {function(Error=)} callback Callback to indicate that the write is @@ -272,6 +283,11 @@ function _write(chunk, encoding, callback) { ServerWritableStream.prototype._write = _write; +/** + * Send the initial metadata for a writable stream. + * @param {Object>} responseMetadata Metadata + * to send + */ function sendMetadata(responseMetadata) { /* jshint validthis: true */ if (!this.call.metadataSent) { @@ -287,6 +303,10 @@ function sendMetadata(responseMetadata) { } } +/** + * @inheritdoc + * @alias module:src/server~ServerWritableStream#sendMetadata + */ ServerWritableStream.prototype.sendMetadata = sendMetadata; util.inherits(ServerReadableStream, Readable); @@ -307,6 +327,7 @@ function ServerReadableStream(call, deserialize) { /** * Start reading from the gRPC data source. This is an implementation of a * method required for implementing stream.Readable + * @access private * @param {number} size Ignored */ function _read(size) { @@ -379,8 +400,22 @@ ServerDuplexStream.prototype._read = _read; ServerDuplexStream.prototype._write = _write; ServerDuplexStream.prototype.sendMetadata = sendMetadata; +/** + * Get the endpoint this call/stream is connected to. + * @return {string} The URI of the endpoint + */ +function getPeer() { + /* jshint validthis: true */ + return this.call.getPeer(); +} + +ServerReadableStream.prototype.getPeer = getPeer; +ServerWritableStream.prototype.getPeer = getPeer; +ServerDuplexStream.prototype.getPeer = getPeer; + /** * Fully handle a unary call + * @access private * @param {grpc.Call} call The call to handle * @param {Object} handler Request handler object for the method that was called * @param {Object} metadata Metadata from the client @@ -395,11 +430,15 @@ function handleUnary(call, handler, metadata) { call.startBatch(batch, function() {}); } }; + emitter.getPeer = function() { + return call.getPeer(); + }; emitter.on('error', function(error) { handleError(call, error); }); emitter.metadata = metadata; waitForCancel(call, emitter); + emitter.call = call; var batch = {}; batch[grpc.opType.RECV_MESSAGE] = true; call.startBatch(batch, function(err, result) { @@ -432,6 +471,7 @@ function handleUnary(call, handler, metadata) { /** * Fully handle a server streaming call + * @access private * @param {grpc.Call} call The call to handle * @param {Object} handler Request handler object for the method that was called * @param {Object} metadata Metadata from the client @@ -460,6 +500,7 @@ function handleServerStreaming(call, handler, metadata) { /** * Fully handle a client streaming call + * @access private * @param {grpc.Call} call The call to handle * @param {Object} handler Request handler object for the method that was called * @param {Object} metadata Metadata from the client @@ -494,6 +535,7 @@ function handleClientStreaming(call, handler, metadata) { /** * Fully handle a bidirectional streaming call + * @access private * @param {grpc.Call} call The call to handle * @param {Object} handler Request handler object for the method that was called * @param {Object} metadata Metadata from the client @@ -550,7 +592,7 @@ function Server(options) { if (err) { return; } - var details = event['new call']; + var details = event.new_call; var call = details.call; var method = details.method; var metadata = details.metadata; @@ -577,7 +619,8 @@ function Server(options) { } server.requestCall(handleNewCall); }; - /** Shuts down the server. + /** + * Shuts down the server. */ this.shutdown = function() { server.shutdown(); @@ -611,6 +654,15 @@ Server.prototype.register = function(name, handler, serialize, deserialize, return true; }; +/** + * Add a service to the server, with a corresponding implementation. If you are + * generating this from a proto file, you should instead use + * addProtoService. + * @param {Object} service The service descriptor, as + * {@link module:src/common.getProtobufServiceAttrs} returns + * @param {Object} implementation Map of method names to + * method implementation for the provided service. + */ Server.prototype.addService = function(service, implementation) { if (this.started) { throw new Error('Can\'t add a service to a started server.'); @@ -648,6 +700,12 @@ Server.prototype.addService = function(service, implementation) { }); }; +/** + * Add a proto service to the server, with a corresponding implementation + * @param {Protobuf.Reflect.Service} service The proto service descriptor + * @param {Object} implementation Map of method names to + * method implementation for the provided service. + */ Server.prototype.addProtoService = function(service, implementation) { this.addService(common.getProtobufServiceAttrs(service), implementation); }; @@ -663,14 +721,10 @@ Server.prototype.bind = function(port, creds) { if (this.started) { throw new Error('Can\'t bind an already running server to an address'); } - if (creds) { - return this._server.addSecureHttp2Port(port, creds); - } else { - return this._server.addHttp2Port(port); - } + return this._server.addHttp2Port(port, creds); }; /** - * See documentation for Server + * @see module:src/server~Server */ exports.Server = Server; diff --git a/test/call_test.js b/test/call_test.js index 98158fff..8d0f20b0 100644 --- a/test/call_test.js +++ b/test/call_test.js @@ -48,14 +48,17 @@ function getDeadline(timeout_secs) { return deadline; } +var insecureCreds = grpc.Credentials.createInsecure(); + describe('call', function() { var channel; var server; before(function() { server = new grpc.Server(); - var port = server.addHttp2Port('localhost:0'); + var port = server.addHttp2Port('localhost:0', + grpc.ServerCredentials.createInsecure()); server.start(); - channel = new grpc.Channel('localhost:' + port); + channel = new grpc.Channel('localhost:' + port, insecureCreds); }); after(function() { server.shutdown(); @@ -81,8 +84,13 @@ describe('call', function() { new grpc.Call(channel, 'method', 0); }); }); + it('should accept an optional fourth string parameter', function() { + assert.doesNotThrow(function() { + new grpc.Call(channel, 'method', new Date(), 'host_override'); + }); + }); it('should fail with a closed channel', function() { - var local_channel = new grpc.Channel('hostname'); + var local_channel = new grpc.Channel('hostname', insecureCreds); local_channel.close(); assert.throws(function() { new grpc.Call(channel, 'method'); @@ -132,7 +140,7 @@ describe('call', function() { 'key2': ['value2']}; call.startBatch(batch, function(err, resp) { assert.ifError(err); - assert.deepEqual(resp, {'send metadata': true}); + assert.deepEqual(resp, {'send_metadata': true}); done(); }); }); @@ -147,7 +155,7 @@ describe('call', function() { }; call.startBatch(batch, function(err, resp) { assert.ifError(err); - assert.deepEqual(resp, {'send metadata': true}); + assert.deepEqual(resp, {'send_metadata': true}); done(); }); }); @@ -184,4 +192,10 @@ describe('call', function() { }); }); }); + describe('getPeer', function() { + it('should return a string', function() { + var call = new grpc.Call(channel, 'method', getDeadline(1)); + assert.strictEqual(typeof call.getPeer(), 'string'); + }); + }); }); diff --git a/test/channel_test.js b/test/channel_test.js index 33200c99..d81df2a3 100644 --- a/test/channel_test.js +++ b/test/channel_test.js @@ -36,11 +36,33 @@ var assert = require('assert'); var grpc = require('bindings')('grpc.node'); +/** + * 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 insecureCreds = grpc.Credentials.createInsecure(); + describe('channel', function() { describe('constructor', function() { it('should require a string for the first argument', function() { assert.doesNotThrow(function() { - new grpc.Channel('hostname'); + new grpc.Channel('hostname', insecureCreds); }); assert.throws(function() { new grpc.Channel(); @@ -49,42 +71,120 @@ describe('channel', function() { new grpc.Channel(5); }); }); - it('should accept an object for the second parameter', function() { + it('should require a credential for the second argument', function() { assert.doesNotThrow(function() { - new grpc.Channel('hostname', {}); + new grpc.Channel('hostname', insecureCreds); }); assert.throws(function() { new grpc.Channel('hostname', 5); }); + assert.throws(function() { + new grpc.Channel('hostname'); + }); + }); + it('should accept an object for the third argument', function() { + assert.doesNotThrow(function() { + new grpc.Channel('hostname', insecureCreds, {}); + }); + assert.throws(function() { + new grpc.Channel('hostname', insecureCreds, 'abc'); + }); }); it('should only accept objects with string or int values', function() { assert.doesNotThrow(function() { - new grpc.Channel('hostname', {'key' : 'value'}); + new grpc.Channel('hostname', insecureCreds,{'key' : 'value'}); }); assert.doesNotThrow(function() { - new grpc.Channel('hostname', {'key' : 5}); + new grpc.Channel('hostname', insecureCreds, {'key' : 5}); }); assert.throws(function() { - new grpc.Channel('hostname', {'key' : null}); + new grpc.Channel('hostname', insecureCreds, {'key' : null}); }); assert.throws(function() { - new grpc.Channel('hostname', {'key' : new Date()}); + new grpc.Channel('hostname', insecureCreds, {'key' : new Date()}); }); }); }); describe('close', function() { + var channel; + beforeEach(function() { + channel = new grpc.Channel('hostname', insecureCreds, {}); + }); it('should succeed silently', function() { - var channel = new grpc.Channel('hostname', {}); assert.doesNotThrow(function() { channel.close(); }); }); it('should be idempotent', function() { - var channel = new grpc.Channel('hostname', {}); assert.doesNotThrow(function() { channel.close(); channel.close(); }); }); }); + describe('getTarget', function() { + var channel; + beforeEach(function() { + channel = new grpc.Channel('hostname', insecureCreds, {}); + }); + it('should return a string', function() { + assert.strictEqual(typeof channel.getTarget(), 'string'); + }); + }); + describe('getConnectivityState', function() { + var channel; + beforeEach(function() { + channel = new grpc.Channel('hostname', insecureCreds, {}); + }); + it('should return IDLE for a new channel', function() { + assert.strictEqual(channel.getConnectivityState(), + grpc.connectivityState.IDLE); + }); + }); + describe('watchConnectivityState', function() { + var channel; + beforeEach(function() { + channel = new grpc.Channel('localhost', insecureCreds, {}); + }); + afterEach(function() { + channel.close(); + }); + it('should time out if called alone', function(done) { + var old_state = channel.getConnectivityState(); + var deadline = new Date(); + deadline.setSeconds(deadline.getSeconds() + 1); + channel.watchConnectivityState(old_state, deadline, function(err, value) { + assert(err); + done(); + }); + }); + it('should complete if a connection attempt is forced', function(done) { + var old_state = channel.getConnectivityState(); + var deadline = new Date(); + deadline.setSeconds(deadline.getSeconds() + 1); + channel.watchConnectivityState(old_state, deadline, function(err, value) { + assert.ifError(err); + assert.notEqual(value.new_state, old_state); + done(); + }); + channel.getConnectivityState(true); + }); + it('should complete twice if called twice', function(done) { + done = multiDone(done, 2); + var old_state = channel.getConnectivityState(); + var deadline = new Date(); + deadline.setSeconds(deadline.getSeconds() + 1); + channel.watchConnectivityState(old_state, deadline, function(err, value) { + assert.ifError(err); + assert.notEqual(value.new_state, old_state); + done(); + }); + channel.watchConnectivityState(old_state, deadline, function(err, value) { + assert.ifError(err); + assert.notEqual(value.new_state, old_state); + done(); + }); + channel.getConnectivityState(true); + }); + }); }); diff --git a/test/constant_test.js b/test/constant_test.js index ecc98ec4..fa06ad4e 100644 --- a/test/constant_test.js +++ b/test/constant_test.js @@ -78,6 +78,31 @@ var callErrorNames = [ 'INVALID_FLAGS' ]; +/** + * 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.} + */ +var connectivityStateNames = [ + 'IDLE', + 'CONNECTING', + 'READY', + 'TRANSIENT_FAILURE', + 'FATAL_FAILURE' +]; + describe('constants', function() { it('should have all of the status constants', function() { for (var i = 0; i < statusNames.length; i++) { @@ -91,4 +116,16 @@ 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]), + 'connectivity status missing: ' + connectivityStateNames[i]); + } + }); }); diff --git a/test/end_to_end_test.js b/test/end_to_end_test.js index 667852f3..7574d98b 100644 --- a/test/end_to_end_test.js +++ b/test/end_to_end_test.js @@ -57,22 +57,23 @@ function multiDone(done, count) { }; } +var insecureCreds = grpc.Credentials.createInsecure(); + describe('end-to-end', function() { var server; var channel; before(function() { server = new grpc.Server(); - var port_num = server.addHttp2Port('0.0.0.0:0'); + var port_num = server.addHttp2Port('0.0.0.0:0', + grpc.ServerCredentials.createInsecure()); server.start(); - channel = new grpc.Channel('localhost:' + port_num); + channel = new grpc.Channel('localhost:' + port_num, insecureCreds); }); after(function() { server.shutdown(); }); it('should start and end a request without error', 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', @@ -85,37 +86,37 @@ describe('end-to-end', function() { call.startBatch(client_batch, function(err, response) { assert.ifError(err); assert.deepEqual(response, { - 'send metadata': true, - 'client close': true, - 'metadata': {}, - 'status': { - 'code': grpc.status.OK, - 'details': status_text, - 'metadata': {} + send_metadata: true, + client_close: true, + metadata: {}, + status: { + code: grpc.status.OK, + details: status_text, + metadata: {} } }); done(); }); server.requestCall(function(err, call_details) { - var new_call = call_details['new call']; + var new_call = call_details.new_call; assert.notEqual(new_call, null); var server_call = new_call.call; assert.notEqual(server_call, null); var server_batch = {}; server_batch[grpc.opType.SEND_INITIAL_METADATA] = {}; server_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = { - 'metadata': {}, - 'code': grpc.status.OK, - 'details': status_text + metadata: {}, + code: grpc.status.OK, + details: status_text }; server_batch[grpc.opType.RECV_CLOSE_ON_SERVER] = true; server_call.startBatch(server_batch, function(err, response) { assert.ifError(err); assert.deepEqual(response, { - 'send metadata': true, - 'send status': true, - 'cancelled': false + send_metadata: true, + send_status: true, + cancelled: false }); done(); }); @@ -123,15 +124,13 @@ describe('end-to-end', function() { }); 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', Infinity); var client_batch = {}; client_batch[grpc.opType.SEND_INITIAL_METADATA] = { - 'client_key': ['client_value'] + client_key: ['client_value'] }; client_batch[grpc.opType.SEND_CLOSE_FROM_CLIENT] = true; client_batch[grpc.opType.RECV_INITIAL_METADATA] = true; @@ -139,18 +138,18 @@ describe('end-to-end', function() { call.startBatch(client_batch, function(err, response) { assert.ifError(err); assert.deepEqual(response,{ - 'send metadata': true, - 'client close': true, + send_metadata: true, + client_close: true, metadata: {server_key: ['server_value']}, - status: {'code': grpc.status.OK, - 'details': status_text, - 'metadata': {}} + status: {code: grpc.status.OK, + details: status_text, + metadata: {}} }); done(); }); server.requestCall(function(err, call_details) { - var new_call = call_details['new call']; + var new_call = call_details.new_call; assert.notEqual(new_call, null); assert.strictEqual(new_call.metadata.client_key[0], 'client_value'); @@ -158,20 +157,20 @@ describe('end-to-end', function() { assert.notEqual(server_call, null); var server_batch = {}; server_batch[grpc.opType.SEND_INITIAL_METADATA] = { - 'server_key': ['server_value'] + server_key: ['server_value'] }; server_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = { - 'metadata': {}, - 'code': grpc.status.OK, - 'details': status_text + metadata: {}, + code: grpc.status.OK, + details: status_text }; server_batch[grpc.opType.RECV_CLOSE_ON_SERVER] = true; server_call.startBatch(server_batch, function(err, response) { assert.ifError(err); assert.deepEqual(response, { - 'send metadata': true, - 'send status': true, - 'cancelled': false + send_metadata: true, + send_status: true, + cancelled: false }); done(); }); @@ -181,8 +180,6 @@ describe('end-to-end', function() { var req_text = 'client_request'; var reply_text = 'server_response'; var done = multiDone(complete, 2); - var deadline = new Date(); - deadline.setSeconds(deadline.getSeconds() + 3); var status_text = 'success'; var call = new grpc.Call(channel, 'dummy_method', @@ -196,19 +193,19 @@ describe('end-to-end', function() { client_batch[grpc.opType.RECV_STATUS_ON_CLIENT] = true; call.startBatch(client_batch, function(err, response) { assert.ifError(err); - assert(response['send metadata']); - assert(response['client close']); + assert(response.send_metadata); + assert(response.client_close); assert.deepEqual(response.metadata, {}); - assert(response['send message']); + assert(response.send_message); assert.strictEqual(response.read.toString(), reply_text); - assert.deepEqual(response.status, {'code': grpc.status.OK, - 'details': status_text, - 'metadata': {}}); + assert.deepEqual(response.status, {code: grpc.status.OK, + details: status_text, + metadata: {}}); done(); }); server.requestCall(function(err, call_details) { - var new_call = call_details['new call']; + var new_call = call_details.new_call; assert.notEqual(new_call, null); var server_call = new_call.call; assert.notEqual(server_call, null); @@ -217,18 +214,18 @@ describe('end-to-end', function() { server_batch[grpc.opType.RECV_MESSAGE] = true; server_call.startBatch(server_batch, function(err, response) { assert.ifError(err); - assert(response['send metadata']); + assert(response.send_metadata); assert.strictEqual(response.read.toString(), req_text); var response_batch = {}; response_batch[grpc.opType.SEND_MESSAGE] = new Buffer(reply_text); response_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = { - 'metadata': {}, - 'code': grpc.status.OK, - 'details': status_text + metadata: {}, + code: grpc.status.OK, + details: status_text }; response_batch[grpc.opType.RECV_CLOSE_ON_SERVER] = true; server_call.startBatch(response_batch, function(err, response) { - assert(response['send status']); + assert(response.send_status); assert(!response.cancelled); done(); }); @@ -238,8 +235,6 @@ describe('end-to-end', function() { it('should send multiple messages', function(complete) { var done = multiDone(complete, 2); var requests = ['req1', 'req2']; - var deadline = new Date(); - deadline.setSeconds(deadline.getSeconds() + 3); var status_text = 'xyz'; var call = new grpc.Call(channel, 'dummy_method', @@ -251,9 +246,9 @@ describe('end-to-end', function() { call.startBatch(client_batch, function(err, response) { assert.ifError(err); assert.deepEqual(response, { - 'send metadata': true, - 'send message': true, - 'metadata': {} + send_metadata: true, + send_message: true, + metadata: {} }); var req2_batch = {}; req2_batch[grpc.opType.SEND_MESSAGE] = new Buffer(requests[1]); @@ -262,12 +257,12 @@ describe('end-to-end', function() { call.startBatch(req2_batch, function(err, resp) { assert.ifError(err); assert.deepEqual(resp, { - 'send message': true, - 'client close': true, - 'status': { - 'code': grpc.status.OK, - 'details': status_text, - 'metadata': {} + send_message: true, + client_close: true, + status: { + code: grpc.status.OK, + details: status_text, + metadata: {} } }); done(); @@ -275,7 +270,7 @@ describe('end-to-end', function() { }); server.requestCall(function(err, call_details) { - var new_call = call_details['new call']; + var new_call = call_details.new_call; assert.notEqual(new_call, null); var server_call = new_call.call; assert.notEqual(server_call, null); @@ -284,7 +279,7 @@ describe('end-to-end', function() { server_batch[grpc.opType.RECV_MESSAGE] = true; server_call.startBatch(server_batch, function(err, response) { assert.ifError(err); - assert(response['send metadata']); + assert(response.send_metadata); assert.strictEqual(response.read.toString(), requests[0]); var snd_batch = {}; snd_batch[grpc.opType.RECV_MESSAGE] = true; @@ -294,13 +289,13 @@ describe('end-to-end', function() { var end_batch = {}; end_batch[grpc.opType.RECV_CLOSE_ON_SERVER] = true; end_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = { - 'metadata': {}, - 'code': grpc.status.OK, - 'details': status_text + metadata: {}, + code: grpc.status.OK, + details: status_text }; server_call.startBatch(end_batch, function(err, response) { assert.ifError(err); - assert(response['send status']); + assert(response.send_status); assert(!response.cancelled); done(); }); diff --git a/test/health_test.js b/test/health_test.js index bb700cc4..be4ef1d2 100644 --- a/test/health_test.js +++ b/test/health_test.js @@ -54,9 +54,11 @@ describe('Health Checking', function() { new health.Implementation(statusMap)); var healthClient; before(function() { - var port_num = healthServer.bind('0.0.0.0:0'); + var port_num = healthServer.bind('0.0.0.0:0', + grpc.ServerCredentials.createInsecure()); healthServer.start(); - healthClient = new health.Client('localhost:' + port_num); + healthClient = new health.Client('localhost:' + port_num, + grpc.Credentials.createInsecure()); }); after(function() { healthServer.shutdown(); diff --git a/test/math_client_test.js b/test/math_client_test.js index f2751857..ef01870a 100644 --- a/test/math_client_test.js +++ b/test/math_client_test.js @@ -51,9 +51,11 @@ var server = require('../examples/math_server.js'); describe('Math client', function() { before(function(done) { - var port_num = server.bind('0.0.0.0:0'); + var port_num = server.bind('0.0.0.0:0', + grpc.ServerCredentials.createInsecure()); server.start(); - math_client = new math.Math('localhost:' + port_num); + math_client = new math.Math('localhost:' + port_num, + grpc.Credentials.createInsecure()); done(); }); after(function() { diff --git a/test/server_test.js b/test/server_test.js index 7cb34fa0..20c9a07f 100644 --- a/test/server_test.js +++ b/test/server_test.js @@ -34,6 +34,8 @@ 'use strict'; var assert = require('assert'); +var fs = require('fs'); +var path = require('path'); var grpc = require('bindings')('grpc.node'); describe('server', function() { @@ -57,7 +59,20 @@ describe('server', function() { it('should bind to an unused port', function() { var port; assert.doesNotThrow(function() { - port = server.addHttp2Port('0.0.0.0:0'); + port = server.addHttp2Port('0.0.0.0:0', + grpc.ServerCredentials.createInsecure()); + }); + assert(port > 0); + }); + it('should bind to an unused port with ssl credentials', function() { + var port; + var key_path = path.join(__dirname, '../test/data/server1.key'); + var pem_path = path.join(__dirname, '../test/data/server1.pem'); + var key_data = fs.readFileSync(key_path); + var pem_data = fs.readFileSync(pem_path); + var creds = grpc.ServerCredentials.createSsl(null, key_data, pem_data); + assert.doesNotThrow(function() { + port = server.addHttp2Port('0.0.0.0:0', creds); }); assert(port > 0); }); @@ -67,25 +82,17 @@ describe('server', function() { before(function() { server = new grpc.Server(); }); - it('should bind to an unused port with fake credentials', function() { - var port; - var creds = grpc.ServerCredentials.createFake(); - assert.doesNotThrow(function() { - port = server.addSecureHttp2Port('0.0.0.0:0', creds); - }); - assert(port > 0); - }); }); - describe('listen', function() { + describe('start', function() { var server; before(function() { server = new grpc.Server(); - server.addHttp2Port('0.0.0.0:0'); + server.addHttp2Port('0.0.0.0:0', grpc.ServerCredentials.createInsecure()); }); after(function() { server.shutdown(); }); - it('should listen without error', function() { + it('should start without error', function() { assert.doesNotThrow(function() { server.start(); }); diff --git a/test/surface_test.js b/test/surface_test.js index 18178e49..52515cc8 100644 --- a/test/surface_test.js +++ b/test/surface_test.js @@ -47,6 +47,29 @@ 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() { it('Should load a proto file by default', function() { assert.doesNotThrow(function() { @@ -110,6 +133,58 @@ describe('Server.prototype.addProtoService', function() { }); }); }); +describe('Client#$waitForReady', function() { + var server; + var port; + var Client; + var client; + before(function() { + server = new grpc.Server(); + port = server.bind('localhost:0', grpc.ServerCredentials.createInsecure()); + server.start(); + Client = surface_client.makeProtobufClientConstructor(mathService); + }); + beforeEach(function() { + client = new Client('localhost:' + port, grpc.Credentials.createInsecure()); + }); + after(function() { + server.shutdown(); + }); + it('should complete when called alone', function(done) { + client.$waitForReady(Infinity, function(error) { + assert.ifError(error); + done(); + }); + }); + it('should complete when a call is initiated', function(done) { + client.$waitForReady(Infinity, function(error) { + assert.ifError(error); + done(); + }); + var call = client.div({}, function(err, response) {}); + call.cancel(); + }); + it('should complete if called more than once', function(done) { + done = multiDone(done, 2); + client.$waitForReady(Infinity, function(error) { + assert.ifError(error); + done(); + }); + client.$waitForReady(Infinity, function(error) { + assert.ifError(error); + done(); + }); + }); + it('should complete if called when already ready', function(done) { + client.$waitForReady(Infinity, function(error) { + assert.ifError(error); + client.$waitForReady(Infinity, function(error) { + assert.ifError(error); + done(); + }); + }); + }); +}); describe('Echo service', function() { var server; var client; @@ -122,9 +197,9 @@ describe('Echo service', function() { callback(null, call.request); } }); - var port = server.bind('localhost:0'); + var port = server.bind('localhost:0', server_insecure_creds); var Client = surface_client.makeProtobufClientConstructor(echo_service); - client = new Client('localhost:' + port); + client = new Client('localhost:' + port, grpc.Credentials.createInsecure()); server.start(); }); after(function() { @@ -166,10 +241,11 @@ describe('Generic client and server', function() { callback(null, _.capitalize(call.request)); } }); - var port = server.bind('localhost:0'); + var port = server.bind('localhost:0', server_insecure_creds); server.start(); var Client = grpc.makeGenericClientConstructor(string_service_attrs); - client = new Client('localhost:' + port); + client = new Client('localhost:' + port, + grpc.Credentials.createInsecure()); }); after(function() { server.shutdown(); @@ -214,9 +290,9 @@ describe('Echo metadata', function() { }); } }); - var port = server.bind('localhost:0'); + var port = server.bind('localhost:0', server_insecure_creds); var Client = surface_client.makeProtobufClientConstructor(test_service); - client = new Client('localhost:' + port); + client = new Client('localhost:' + port, grpc.Credentials.createInsecure()); server.start(); }); after(function() { @@ -258,14 +334,25 @@ describe('Echo metadata', function() { }); call.end(); }); + it('shows the correct user-agent string', function(done) { + var version = require('../package.json').version; + var call = client.unary({}, function(err, data) { assert.ifError(err); }, + {key: ['value']}); + call.on('metadata', function(metadata) { + assert(_.startsWith(metadata['user-agent'], 'grpc-node/' + version)); + done(); + }); + }); }); 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) { @@ -326,14 +413,17 @@ describe('Other conditions', function() { }); } }); - port = server.bind('localhost:0'); - var Client = surface_client.makeProtobufClientConstructor(test_service); - client = new Client('localhost:' + port); + port = server.bind('localhost:0', server_insecure_creds); + Client = surface_client.makeProtobufClientConstructor(test_service); + client = new Client('localhost:' + port, grpc.Credentials.createInsecure()); server.start(); }); after(function() { server.shutdown(); }); + it('channel.getTarget should be available', function() { + assert.strictEqual(typeof client.channel.getTarget(), 'string'); + }); describe('Server recieving bad input', function() { var misbehavingClient; var badArg = new Buffer([0xFF]); @@ -370,7 +460,8 @@ describe('Other conditions', function() { }; var Client = surface_client.makeClientConstructor(test_service_attrs, 'TestService'); - misbehavingClient = new Client('localhost:' + port); + misbehavingClient = new Client('localhost:' + port, + grpc.Credentials.createInsecure()); }); it('should respond correctly to a unary call', function(done) { misbehavingClient.unary(badArg, function(err, data) { @@ -539,6 +630,203 @@ describe('Other conditions', function() { }); }); }); + describe('call.getPeer should return the peer', function() { + it('for a unary call', function(done) { + var call = client.unary({error: false}, function(err, data) { + assert.ifError(err); + done(); + }); + assert.strictEqual(typeof call.getPeer(), 'string'); + }); + it('for a client stream call', function(done) { + var call = client.clientStream(function(err, data) { + assert.ifError(err); + done(); + }); + assert.strictEqual(typeof call.getPeer(), 'string'); + call.write({error: false}); + call.end(); + }); + it('for a server stream call', function(done) { + var call = client.serverStream({error: false}); + assert.strictEqual(typeof call.getPeer(), 'string'); + call.on('data', function(){}); + call.on('status', function(status) { + assert.strictEqual(status.code, grpc.status.OK); + done(); + }); + }); + it('for a bidi stream call', function(done) { + var call = client.bidiStream(); + assert.strictEqual(typeof call.getPeer(), 'string'); + call.write({error: false}); + call.end(); + call.on('data', function(){}); + call.on('status', function(status) { + done(); + }); + }); + }); + 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; @@ -551,9 +839,9 @@ describe('Cancelling surface client', function() { 'fib': function(stream) {}, 'sum': function(stream) {} }); - var port = server.bind('localhost:0'); + var port = server.bind('localhost:0', server_insecure_creds); var Client = surface_client.makeProtobufClientConstructor(mathService); - client = new Client('localhost:' + port); + client = new Client('localhost:' + port, grpc.Credentials.createInsecure()); server.start(); }); after(function() {