grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Resolved merge conflicts with master

This commit is contained in:
murgatroid99 2015-08-17 13:52:09 -07:00
parent 146c61aeec 2cbaf1e9ac
commit fb34cbe5dc
38 changed files with 1099 additions and 384 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -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

3
binding.gyp
View File

@ -11,7 +11,8 @@
'-pedantic',
'-g',
'-zdefs',
'-Werror'
'-Werror',
'-Wno-error=deprecated-declarations'
],
'ldflags': [
'-g'

2
examples/math_server.js
View File

@ -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();
}

7
examples/perf_test.js
View File

@ -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);

5
examples/qps_test.js
View File

@ -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;

3
examples/route_guide_client.js
View File

@ -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;

4
examples/route_guide_server.js
View File

@ -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();
});
}

3
examples/stock_client.js
View File

@ -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);
* });

4
examples/stock_server.js
View File

@ -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;

60
ext/call.cc
View File

@ -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<Object> exports) {
NanNew<FunctionTemplate>(StartBatch)->GetFunction());
NanSetPrototypeTemplate(tpl, "cancel",
NanNew<FunctionTemplate>(Cancel)->GetFunction());
NanSetPrototypeTemplate(tpl, "getPeer",
NanNew<FunctionTemplate>(GetPeer)->GetFunction());
NanAssignPersistent(fun_tpl, tpl);
Handle<Function> 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<Call>(args[4]->ToObject());
parent_call = parent_obj->wrapped_call;
} else if (!(args[4]->IsUndefined() || args[4]->IsNull())) {
return NanThrowTypeError(
"Call's fifth argument must be another call, if provided");
}
gpr_uint32 propagate_flags = GRPC_PROPAGATE_DEFAULTS;
if (args[5]->IsUint32()) {
propagate_flags = args[5]->Uint32Value();
} else if (!(args[5]->IsUndefined() || args[5]->IsNull())) {
return NanThrowTypeError(
"Call's sixth argument must be propagate flags, if provided");
}
Handle<Object> channel_object = args[0]->ToObject();
Channel *channel = ObjectWrap::Unwrap<Channel>(channel_object);
if (channel->GetWrappedChannel() == NULL) {
@ -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<Call>(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<Call>(args.This());
char *peer = grpc_call_get_peer(call->wrapped_call);
Handle<Value> peer_value = NanNew(peer);
gpr_free(peer);
NanReturnValue(peer_value);
}
} // namespace node
} // namespace grpc

1
ext/call.h
View File

@ -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<v8::FunctionTemplate> fun_tpl;

134
ext/channel.cc
View File

@ -33,12 +33,17 @@
#include <vector>
#include "grpc/support/log.h"
#include <node.h>
#include <nan.h>
#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<FunctionTemplate> 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<Object> exports) {
@ -76,6 +80,14 @@ void Channel::Init(Handle<Object> exports) {
tpl->InstanceTemplate()->SetInternalFieldCount(1);
NanSetPrototypeTemplate(tpl, "close",
NanNew<FunctionTemplate>(Close)->GetFunction());
NanSetPrototypeTemplate(tpl, "getTarget",
NanNew<FunctionTemplate>(GetTarget)->GetFunction());
NanSetPrototypeTemplate(
tpl, "getConnectivityState",
NanNew<FunctionTemplate>(GetConnectivityState)->GetFunction());
NanSetPrototypeTemplate(
tpl, "watchConnectivityState",
NanNew<FunctionTemplate>(WatchConnectivityState)->GetFunction());
NanAssignPersistent(fun_tpl, tpl);
Handle<Function> ctr = tpl->GetFunction();
constructor = new NanCallback(ctr);
@ -89,38 +101,31 @@ bool Channel::HasInstance(Handle<Value> 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<Object> 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<Value> creds_value = args_hash->Get(NanNew("credentials"));
if (!Credentials::HasInstance(creds_value)) {
NanUtf8String host(args[0]);
grpc_credentials *creds;
if (!Credentials::HasInstance(args[1])) {
return NanThrowTypeError(
"credentials arg must be a Credentials object");
"Channel's second argument must be a credential");
}
Credentials *creds_object =
ObjectWrap::Unwrap<Credentials>(creds_value->ToObject());
Credentials *creds_object = ObjectWrap::Unwrap<Credentials>(
args[1]->ToObject());
creds = creds_object->GetWrappedCredentials();
args_hash->Delete(NanNew("credentials"));
}
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<Object> args_hash(args[2]->ToObject()->Clone());
Handle<Array> 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<Value> argv[argc] = {args[0], args[1]};
const int argc = 3;
Local<Value> 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<Channel>(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<Channel>(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<grpc_connectivity_state>(args[0]->Uint32Value());
double deadline = args[1]->NumberValue();
Handle<Function> callback_func = args[2].As<Function>();
NanCallback *callback = new NanCallback(callback_func);
Channel *channel = ObjectWrap::Unwrap<Channel>(args.This());
unique_ptr<OpVec> 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<Resources>(nullptr)));
CompletionQueueAsyncWorker::Next();
NanReturnUndefined();
}
} // namespace node
} // namespace grpc

9
ext/channel.h
View File

@ -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<v8::FunctionTemplate> fun_tpl;
grpc_channel *wrapped_channel;
NanUtf8String *host;
};
} // namespace node

6
ext/completion_queue_async_worker.cc
View File

@ -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<Object> exports) {
NanScope();
current_threads = 0;
waiting_next_calls = 0;
queue = grpc_completion_queue_create();
queue = grpc_completion_queue_create(NULL);
}
void CompletionQueueAsyncWorker::HandleOKCallback() {

52
ext/credentials.cc
View File

@ -79,10 +79,10 @@ void Credentials::Init(Handle<Object> exports) {
NanNew<FunctionTemplate>(CreateComposite)->GetFunction());
ctr->Set(NanNew("createGce"),
NanNew<FunctionTemplate>(CreateGce)->GetFunction());
ctr->Set(NanNew("createFake"),
NanNew<FunctionTemplate>(CreateFake)->GetFunction());
ctr->Set(NanNew("createIam"),
NanNew<FunctionTemplate>(CreateIam)->GetFunction());
ctr->Set(NanNew("createInsecure"),
NanNew<FunctionTemplate>(CreateInsecure)->GetFunction());
constructor = new NanCallback(ctr);
exports->Set(NanNew("Credentials"), ctr);
}
@ -94,9 +94,6 @@ bool Credentials::HasInstance(Handle<Value> val) {
Handle<Value> Credentials::WrapStruct(grpc_credentials *credentials) {
NanEscapableScope();
if (credentials == NULL) {
return NanEscapeScope(NanNull());
}
const int argc = 1;
Handle<Value> argv[argc] = {
NanNew<External>(reinterpret_cast<void *>(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<Credentials>(args[0]->ToObject());
Credentials *creds2 = ObjectWrap::Unwrap<Credentials>(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()));
grpc_credentials *creds = grpc_compute_engine_credentials_create();
if (creds == NULL) {
NanReturnNull();
}
NAN_METHOD(Credentials::CreateFake) {
NanScope();
NanReturnValue(WrapStruct(grpc_fake_transport_security_credentials_create()));
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

1
ext/credentials.h
View File

@ -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<v8::FunctionTemplate> fun_tpl;

39
ext/node_grpc.cc
View File

@ -159,12 +159,51 @@ void InitOpTypeConstants(Handle<Object> exports) {
op_type->Set(NanNew("RECV_CLOSE_ON_SERVER"), RECV_CLOSE_ON_SERVER);
}
void InitPropagateConstants(Handle<Object> exports) {
NanScope();
Handle<Object> propagate = NanNew<Object>();
exports->Set(NanNew("propagate"), propagate);
Handle<Value> DEADLINE(NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_DEADLINE));
propagate->Set(NanNew("DEADLINE"), DEADLINE);
Handle<Value> CENSUS_STATS_CONTEXT(
NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_CENSUS_STATS_CONTEXT));
propagate->Set(NanNew("CENSUS_STATS_CONTEXT"), CENSUS_STATS_CONTEXT);
Handle<Value> CENSUS_TRACING_CONTEXT(
NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_CENSUS_TRACING_CONTEXT));
propagate->Set(NanNew("CENSUS_TRACING_CONTEXT"), CENSUS_TRACING_CONTEXT);
Handle<Value> CANCELLATION(
NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_CANCELLATION));
propagate->Set(NanNew("CANCELLATION"), CANCELLATION);
Handle<Value> DEFAULTS(NanNew<Uint32, uint32_t>(GRPC_PROPAGATE_DEFAULTS));
propagate->Set(NanNew("DEFAULTS"), DEFAULTS);
}
void InitConnectivityStateConstants(Handle<Object> exports) {
NanScope();
Handle<Object> channel_state = NanNew<Object>();
exports->Set(NanNew("connectivityState"), channel_state);
Handle<Value> IDLE(NanNew<Uint32, uint32_t>(GRPC_CHANNEL_IDLE));
channel_state->Set(NanNew("IDLE"), IDLE);
Handle<Value> CONNECTING(NanNew<Uint32, uint32_t>(GRPC_CHANNEL_CONNECTING));
channel_state->Set(NanNew("CONNECTING"), CONNECTING);
Handle<Value> READY(NanNew<Uint32, uint32_t>(GRPC_CHANNEL_READY));
channel_state->Set(NanNew("READY"), READY);
Handle<Value> TRANSIENT_FAILURE(
NanNew<Uint32, uint32_t>(GRPC_CHANNEL_TRANSIENT_FAILURE));
channel_state->Set(NanNew("TRANSIENT_FAILURE"), TRANSIENT_FAILURE);
Handle<Value> FATAL_FAILURE(
NanNew<Uint32, uint32_t>(GRPC_CHANNEL_FATAL_FAILURE));
channel_state->Set(NanNew("FATAL_FAILURE"), FATAL_FAILURE);
}
void init(Handle<Object> exports) {
NanScope();
grpc_init();
InitStatusConstants(exports);
InitCallErrorConstants(exports);
InitOpTypeConstants(exports);
InitPropagateConstants(exports);
InitConnectivityStateConstants(exports);
grpc::node::Call::Init(exports);
grpc::node::Channel::Init(exports);

58
ext/server.cc
View File

@ -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<Object> exports) {
tpl, "addHttp2Port",
NanNew<FunctionTemplate>(AddHttp2Port)->GetFunction());
NanSetPrototypeTemplate(
tpl, "addSecureHttp2Port",
NanNew<FunctionTemplate>(AddSecureHttp2Port)->GetFunction());
NanSetPrototypeTemplate(tpl, "start",
NanNew<FunctionTemplate>(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<Object> args_hash(args[0]->ToObject());
Handle<Array> 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<Server>(args.This());
if (server->wrapped_server == NULL) {
return NanThrowError("addHttp2Port cannot be called on a shut down Server");
}
NanReturnValue(NanNew<Number>(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<Server>(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>(
ServerCredentials *creds_object = ObjectWrap::Unwrap<ServerCredentials>(
args[1]->ToObject());
NanReturnValue(NanNew<Number>(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<Number>(port));
}
NAN_METHOD(Server::Start) {

1
ext/server.h
View File

@ -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;

22
ext/server_credentials.cc
View File

@ -73,8 +73,8 @@ void ServerCredentials::Init(Handle<Object> exports) {
Handle<Function> ctr = tpl->GetFunction();
ctr->Set(NanNew("createSsl"),
NanNew<FunctionTemplate>(CreateSsl)->GetFunction());
ctr->Set(NanNew("createFake"),
NanNew<FunctionTemplate>(CreateFake)->GetFunction());
ctr->Set(NanNew("createInsecure"),
NanNew<FunctionTemplate>(CreateInsecure)->GetFunction());
constructor = new NanCallback(ctr);
exports->Set(NanNew("ServerCredentials"), ctr);
}
@ -87,9 +87,6 @@ bool ServerCredentials::HasInstance(Handle<Value> val) {
Handle<Value> ServerCredentials::WrapStruct(
grpc_server_credentials *credentials) {
NanEscapableScope();
if (credentials == NULL) {
return NanEscapeScope(NanNull());
}
const int argc = 1;
Handle<Value> argv[argc] = {
NanNew<External>(reinterpret_cast<void *>(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

2
ext/server_credentials.h
View File

@ -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<v8::FunctionTemplate> fun_tpl;

1
ext/timeval.cc
View File

@ -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<double>::infinity();
} else if (gpr_time_cmp(timespec, gpr_inf_past(GPR_CLOCK_REALTIME)) == 0) {

37
index.js
View File

@ -48,7 +48,7 @@ var grpc = require('bindings')('grpc');
* @param {ProtoBuf.Reflect.Namespace} value The ProtoBuf object to load.
* @return {Object<string, *>} 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<string, *>} 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;

39
interop/interop_client.js
View File

@ -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);
if (scope) {
assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE);
});
call.on('status', function(status) {
assert.strictEqual(status.code, grpc.status.OK);
}
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);
}

6
interop/interop_server.js
View File

@ -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();
}
/**

22
jsdoc_conf.json Normal file
View File

@ -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
}
}
}

4
package.json
View File

@ -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",

163
src/client.js
View File

@ -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
*/

52
src/common.js
View File

@ -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;
};

70
src/server.js
View File

@ -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<String, Array<(String|Buffer)>>} 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<String, *>} service The service descriptor, as
* {@link module:src/common.getProtobufServiceAttrs} returns
* @param {Object<String, function>} 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<String, function>} 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;

24
test/call_test.js
View File

@ -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');
});
});
});

118
test/channel_test.js
View File

@ -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);
});
});
});

37
test/constant_test.js
View File

@ -78,6 +78,31 @@ var callErrorNames = [
'INVALID_FLAGS'
];
/**
* List of all propagate flag names
* @const
* @type {Array.<string>}
*/
var propagateFlagNames = [
'DEADLINE',
'CENSUS_STATS_CONTEXT',
'CENSUS_TRACING_CONTEXT',
'CANCELLATION',
'DEFAULTS'
];
/*
* List of all connectivity state names
* @const
* @type {Array.<string>}
*/
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]);
}
});
});

125
test/end_to_end_test.js
View File

@ -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();
});

6
test/health_test.js
View File

@ -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();

6
test/math_client_test.js
View File

@ -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() {

31
test/server_test.js
View File

@ -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();
});

314
test/surface_test.js
View File

@ -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() {