diff options
author | Alistair Veitch <aveitch@google.com> | 2015-07-26 15:24:41 -0700 |
---|---|---|
committer | Alistair Veitch <aveitch@google.com> | 2015-07-26 15:24:41 -0700 |
commit | b4cbc1e2f72b18da3c84130c5bf7b28e344fba8f (patch) | |
tree | 5564c6754f30570bb59c6e62c2ddda405a51296b /src/node | |
parent | af5002f9ae647f8d82ec3b1cdaef4438cd6d2ad0 (diff) | |
parent | 5c575dd6e4b01cd68cca5d1917b58023dcf4ca0f (diff) |
post-merge
Diffstat (limited to 'src/node')
27 files changed, 586 insertions, 397 deletions
diff --git a/src/node/README.md b/src/node/README.md index 2f4c49096d..7d3d8c7fa1 100644 --- a/src/node/README.md +++ b/src/node/README.md @@ -54,10 +54,10 @@ function loadObject(reflectionObject) Returns the same structure that `load` returns, but takes a reflection object from `ProtoBuf.js` instead of a file name. ```javascript -function buildServer(serviceArray) +function Server([serverOpions]) ``` -Takes an array of service objects and returns a constructor for a server that handles requests to all of those services. +Constructs a server to which service/implementation pairs can be added. ```javascript @@ -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/src/node/examples/math_server.js b/src/node/examples/math_server.js index 0a86e7eaff..b1f8a6323f 100644 --- a/src/node/examples/math_server.js +++ b/src/node/examples/math_server.js @@ -36,8 +36,6 @@ var grpc = require('..'); var math = grpc.load(__dirname + '/math.proto').math; -var Server = grpc.buildServer([math.Math.service]); - /** * Server function for division. Provides the /Math/DivMany and /Math/Div * functions (Div is just DivMany with only one stream element). For each @@ -108,19 +106,17 @@ function mathDivMany(stream) { stream.end(); }); } - -var server = new Server({ - 'math.Math' : { - div: mathDiv, - fib: mathFib, - sum: mathSum, - divMany: mathDivMany - } +var server = new grpc.Server(); +server.addProtoService(math.Math.service, { + div: mathDiv, + fib: mathFib, + sum: mathSum, + divMany: mathDivMany }); if (require.main === module) { server.bind('0.0.0.0:50051'); - server.listen(); + server.start(); } /** diff --git a/src/node/examples/route_guide_server.js b/src/node/examples/route_guide_server.js index c777eab7bc..70044a322c 100644 --- a/src/node/examples/route_guide_server.js +++ b/src/node/examples/route_guide_server.js @@ -40,8 +40,6 @@ var _ = require('lodash'); var grpc = require('..'); var examples = grpc.load(__dirname + '/route_guide.proto').examples; -var Server = grpc.buildServer([examples.RouteGuide.service]); - var COORD_FACTOR = 1e7; /** @@ -228,14 +226,14 @@ function routeChat(call) { * @return {Server} The new server object */ function getServer() { - return new Server({ - 'examples.RouteGuide' : { - getFeature: getFeature, - listFeatures: listFeatures, - recordRoute: recordRoute, - routeChat: routeChat - } + var server = new grpc.Server(); + server.addProtoService(examples.RouteGuide.service, { + getFeature: getFeature, + listFeatures: listFeatures, + recordRoute: recordRoute, + routeChat: routeChat }); + return server; } if (require.main === module) { diff --git a/src/node/examples/stock_server.js b/src/node/examples/stock_server.js index caaf9f99ba..f2eb6ad4ab 100644 --- a/src/node/examples/stock_server.js +++ b/src/node/examples/stock_server.js @@ -37,8 +37,6 @@ var _ = require('lodash'); var grpc = require('..'); var examples = grpc.load(__dirname + '/stock.proto').examples; -var StockServer = grpc.buildServer([examples.Stock.service]); - function getLastTradePrice(call, callback) { callback(null, {symbol: call.request.symbol, price: 88}); } @@ -73,13 +71,12 @@ function getLastTradePriceMultiple(call) { }); } -var stockServer = new StockServer({ - 'examples.Stock' : { - getLastTradePrice: getLastTradePrice, - getLastTradePriceMultiple: getLastTradePriceMultiple, - watchFutureTrades: watchFutureTrades, - getHighestTradePrice: getHighestTradePrice - } +var stockServer = new grpc.Server(); +stockServer.addProtoService(examples.Stock.service, { + getLastTradePrice: getLastTradePrice, + getLastTradePriceMultiple: getLastTradePriceMultiple, + watchFutureTrades: watchFutureTrades, + getHighestTradePrice: getHighestTradePrice }); if (require.main === module) { diff --git a/src/node/ext/call.cc b/src/node/ext/call.cc index 15c9b2d97d..dc45c8d8ae 100644 --- a/src/node/ext/call.cc +++ b/src/node/ext/call.cc @@ -192,7 +192,7 @@ class SendMetadataOp : public Op { } protected: std::string GetTypeString() const { - return "send metadata"; + return "send_metadata"; } }; @@ -216,7 +216,7 @@ class SendMessageOp : public Op { } protected: std::string GetTypeString() const { - return "send message"; + return "send_message"; } }; @@ -232,7 +232,7 @@ class SendClientCloseOp : public Op { } protected: std::string GetTypeString() const { - return "client close"; + return "client_close"; } }; @@ -276,7 +276,7 @@ class SendServerStatusOp : public Op { } protected: std::string GetTypeString() const { - return "send status"; + return "send_status"; } }; @@ -453,6 +453,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"), @@ -608,5 +610,17 @@ NAN_METHOD(Call::Cancel) { 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 diff --git a/src/node/ext/call.h b/src/node/ext/call.h index 43142c7091..6acda76197 100644 --- a/src/node/ext/call.h +++ b/src/node/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<v8::FunctionTemplate> fun_tpl; diff --git a/src/node/ext/channel.cc b/src/node/ext/channel.cc index d37bf763dd..0b7333e450 100644 --- a/src/node/ext/channel.cc +++ b/src/node/ext/channel.cc @@ -76,6 +76,8 @@ void Channel::Init(Handle<Object> exports) { tpl->InstanceTemplate()->SetInternalFieldCount(1); NanSetPrototypeTemplate(tpl, "close", NanNew<FunctionTemplate>(Close)->GetFunction()); + NanSetPrototypeTemplate(tpl, "getTarget", + NanNew<FunctionTemplate>(GetTarget)->GetFunction()); NanAssignPersistent(fun_tpl, tpl); Handle<Function> ctr = tpl->GetFunction(); constructor = new NanCallback(ctr); @@ -185,5 +187,14 @@ 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))); +} + } // namespace node } // namespace grpc diff --git a/src/node/ext/channel.h b/src/node/ext/channel.h index b3aa0f700f..6725ebb03f 100644 --- a/src/node/ext/channel.h +++ b/src/node/ext/channel.h @@ -66,6 +66,7 @@ class Channel : public ::node::ObjectWrap { static NAN_METHOD(New); static NAN_METHOD(Close); + static NAN_METHOD(GetTarget); static NanCallback *constructor; static v8::Persistent<v8::FunctionTemplate> fun_tpl; diff --git a/src/node/ext/completion_queue_async_worker.cc b/src/node/ext/completion_queue_async_worker.cc index 4be208c82d..1215c97e19 100644 --- a/src/node/ext/completion_queue_async_worker.cc +++ b/src/node/ext/completion_queue_async_worker.cc @@ -62,7 +62,8 @@ CompletionQueueAsyncWorker::CompletionQueueAsyncWorker() CompletionQueueAsyncWorker::~CompletionQueueAsyncWorker() {} void CompletionQueueAsyncWorker::Execute() { - result = grpc_completion_queue_next(queue, gpr_inf_future); + result = + grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME)); if (!result.success) { SetErrorMessage("The batch encountered an error"); } diff --git a/src/node/ext/credentials.cc b/src/node/ext/credentials.cc index 34872017ea..d6cff0631d 100644 --- a/src/node/ext/credentials.cc +++ b/src/node/ext/credentials.cc @@ -79,8 +79,6 @@ 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()); constructor = new NanCallback(ctr); @@ -180,11 +178,6 @@ NAN_METHOD(Credentials::CreateGce) { NanReturnValue(WrapStruct(grpc_compute_engine_credentials_create())); } -NAN_METHOD(Credentials::CreateFake) { - NanScope(); - NanReturnValue(WrapStruct(grpc_fake_transport_security_credentials_create())); -} - NAN_METHOD(Credentials::CreateIam) { NanScope(); if (!args[0]->IsString()) { diff --git a/src/node/ext/server.cc b/src/node/ext/server.cc index 51c55ba965..8554fce777 100644 --- a/src/node/ext/server.cc +++ b/src/node/ext/server.cc @@ -108,7 +108,7 @@ class NewCallOp : public Op { protected: std::string GetTypeString() const { - return "new call"; + return "new_call"; } }; @@ -161,7 +161,8 @@ void Server::ShutdownServer() { grpc_server_shutdown_and_notify(this->wrapped_server, this->shutdown_queue, NULL); - grpc_completion_queue_pluck(this->shutdown_queue, NULL, gpr_inf_future); + grpc_completion_queue_pluck(this->shutdown_queue, NULL, + gpr_inf_future(GPR_CLOCK_REALTIME)); this->wrapped_server = NULL; } } diff --git a/src/node/ext/server_credentials.cc b/src/node/ext/server_credentials.cc index d2b63cdc4e..66aaa3300f 100644 --- a/src/node/ext/server_credentials.cc +++ b/src/node/ext/server_credentials.cc @@ -73,8 +73,6 @@ 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()); constructor = new NanCallback(ctr); exports->Set(NanNew("ServerCredentials"), ctr); } @@ -144,11 +142,5 @@ NAN_METHOD(ServerCredentials::CreateSsl) { grpc_ssl_server_credentials_create(root_certs, &key_cert_pair, 1))); } -NAN_METHOD(ServerCredentials::CreateFake) { - NanScope(); - NanReturnValue( - WrapStruct(grpc_fake_transport_security_server_credentials_create())); -} - } // namespace node } // namespace grpc diff --git a/src/node/ext/server_credentials.h b/src/node/ext/server_credentials.h index aaa7ef297a..80747504a1 100644 --- a/src/node/ext/server_credentials.h +++ b/src/node/ext/server_credentials.h @@ -63,7 +63,6 @@ class ServerCredentials : public ::node::ObjectWrap { static NAN_METHOD(New); static NAN_METHOD(CreateSsl); - static NAN_METHOD(CreateFake); static NanCallback *constructor; // Used for typechecking instances of this javascript class static v8::Persistent<v8::FunctionTemplate> fun_tpl; diff --git a/src/node/ext/timeval.cc b/src/node/ext/timeval.cc index bc3237f7a6..bf68513c48 100644 --- a/src/node/ext/timeval.cc +++ b/src/node/ext/timeval.cc @@ -42,18 +42,20 @@ namespace node { gpr_timespec MillisecondsToTimespec(double millis) { if (millis == std::numeric_limits<double>::infinity()) { - return gpr_inf_future; + return gpr_inf_future(GPR_CLOCK_REALTIME); } else if (millis == -std::numeric_limits<double>::infinity()) { - return gpr_inf_past; + return gpr_inf_past(GPR_CLOCK_REALTIME); } else { - return gpr_time_from_micros(static_cast<int64_t>(millis * 1000)); + return gpr_time_from_micros(static_cast<int64_t>(millis * 1000), + GPR_CLOCK_REALTIME); } } double TimespecToMilliseconds(gpr_timespec timespec) { - if (gpr_time_cmp(timespec, gpr_inf_future) == 0) { + 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) == 0) { + } else if (gpr_time_cmp(timespec, gpr_inf_past(GPR_CLOCK_REALTIME)) == 0) { return -std::numeric_limits<double>::infinity(); } else { return (static_cast<double>(timespec.tv_sec) * 1000 + diff --git a/src/node/index.js b/src/node/index.js index b6a4e2d0ee..d81e780443 100644 --- a/src/node/index.js +++ b/src/node/index.js @@ -133,9 +133,9 @@ exports.loadObject = loadObject; exports.load = load; /** - * See docs for server.makeServerConstructor + * See docs for Server */ -exports.buildServer = server.makeProtobufServerConstructor; +exports.Server = server.Server; /** * Status name to code number mapping @@ -159,5 +159,3 @@ exports.ServerCredentials = grpc.ServerCredentials; exports.getGoogleAuthDelegate = getGoogleAuthDelegate; exports.makeGenericClientConstructor = client.makeClientConstructor; - -exports.makeGenericServerConstructor = server.makeServerConstructor; diff --git a/src/node/interop/interop_client.js b/src/node/interop/interop_client.js index b61b0b63c0..e810e68e45 100644 --- a/src/node/interop/interop_client.js +++ b/src/node/interop/interop_client.js @@ -318,6 +318,51 @@ function authTest(expected_user, scope, client, done) { }); } +function oauth2Test(expected_user, scope, per_rpc, client, done) { + (new GoogleAuth()).getApplicationDefault(function(err, credential) { + assert.ifError(err); + var arg = { + fill_username: true, + fill_oauth_scope: true + }; + credential = credential.createScoped(scope); + credential.getAccessToken(function(err, token) { + assert.ifError(err); + var updateMetadata = function(authURI, metadata, callback) { + metadata = _.clone(metadata); + if (metadata.Authorization) { + metadata.Authorization = _.clone(metadata.Authorization); + } else { + metadata.Authorization = []; + } + metadata.Authorization.push('Bearer ' + token); + callback(null, metadata); + }; + var makeTestCall = function(error, client_metadata) { + assert.ifError(error); + var call = 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(); + } + }); + }; + if (per_rpc) { + updateMetadata('', {}, makeTestCall); + } else { + client.updateMetadata = updateMetadata; + makeTestCall(null, {}); + } + + }); + }); +} + /** * Map from test case names to test functions */ @@ -333,7 +378,9 @@ var test_cases = { timeout_on_sleeping_server: timeoutOnSleepingServer, compute_engine_creds: _.partial(authTest, COMPUTE_ENGINE_USER, null), service_account_creds: _.partial(authTest, AUTH_USER, AUTH_SCOPE), - jwt_token_creds: _.partial(authTest, AUTH_USER, null) + jwt_token_creds: _.partial(authTest, AUTH_USER, null), + oauth2_auth_token: _.partial(oauth2Test, AUTH_USER, AUTH_SCOPE, false), + per_rpc_creds: _.partial(oauth2Test, AUTH_USER, AUTH_SCOPE, true) }; /** diff --git a/src/node/interop/interop_server.js b/src/node/interop/interop_server.js index 0baa78a094..505c6bb537 100644 --- a/src/node/interop/interop_server.js +++ b/src/node/interop/interop_server.js @@ -38,7 +38,6 @@ var path = require('path'); var _ = require('lodash'); var grpc = require('..'); var testProto = grpc.load(__dirname + '/test.proto').grpc.testing; -var Server = grpc.buildServer([testProto.TestService.service]); /** * Create a buffer filled with size zeroes @@ -173,16 +172,15 @@ function getServer(port, tls) { key_data, pem_data); } - var server = new Server({ - 'grpc.testing.TestService' : { - emptyCall: handleEmpty, - unaryCall: handleUnary, - streamingOutputCall: handleStreamingOutput, - streamingInputCall: handleStreamingInput, - fullDuplexCall: handleFullDuplex, - halfDuplexCall: handleHalfDuplex - } - }, null, options); + var server = new grpc.Server(options); + server.addProtoService(testProto.TestService.service, { + emptyCall: handleEmpty, + unaryCall: handleUnary, + streamingOutputCall: handleStreamingOutput, + streamingInputCall: handleStreamingInput, + fullDuplexCall: handleFullDuplex, + halfDuplexCall: handleHalfDuplex + }); var port_num = server.bind('0.0.0.0:' + port, server_creds); return {server: server, port: port_num}; } diff --git a/src/node/src/client.js b/src/node/src/client.js index b7bad949d4..d89c656c07 100644 --- a/src/node/src/client.js +++ b/src/node/src/client.js @@ -47,6 +47,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); @@ -187,6 +188,19 @@ 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 function that can make unary requests to the specified method. * @param {string} method The name of the method to request * @param {function(*):Buffer} serialize The serialization function for inputs @@ -222,6 +236,9 @@ function makeUnaryRequestFunction(method, serialize, deserialize) { 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(); @@ -517,9 +534,12 @@ function makeClientConstructor(methods, serviceName) { callback(null, metadata); }; } - - this.server_address = address.replace(/\/$/, ''); + if (!options) { + options = {}; + } + options['grpc.primary_user_agent'] = 'grpc-node/' + version; this.channel = new grpc.Channel(address, options); + this.server_address = address.replace(/\/$/, ''); this.auth_uri = this.server_address + '/' + serviceName; this.updateMetadata = updateMetadata; } diff --git a/src/node/src/server.js b/src/node/src/server.js index 00be400e61..e876313d96 100644 --- a/src/node/src/server.js +++ b/src/node/src/server.js @@ -72,6 +72,9 @@ function handleError(call, error) { status.metadata = error.metadata; } var error_batch = {}; + if (!call.metadataSent) { + error_batch[grpc.opType.SEND_INITIAL_METADATA] = {}; + } error_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = status; call.startBatch(error_batch, function(){}); } @@ -115,6 +118,10 @@ function sendUnaryResponse(call, value, serialize, metadata) { if (metadata) { status.metadata = metadata; } + if (!call.metadataSent) { + end_batch[grpc.opType.SEND_INITIAL_METADATA] = {}; + call.metadataSent = true; + } end_batch[grpc.opType.SEND_MESSAGE] = serialize(value); end_batch[grpc.opType.SEND_STATUS_FROM_SERVER] = status; call.startBatch(end_batch, function (){}); @@ -136,6 +143,10 @@ function setUpWritable(stream, serialize) { stream.serialize = common.wrapIgnoreNull(serialize); function sendStatus() { var batch = {}; + if (!stream.call.metadataSent) { + stream.call.metadataSent = true; + batch[grpc.opType.SEND_INITIAL_METADATA] = {}; + } batch[grpc.opType.SEND_STATUS_FROM_SERVER] = stream.status; stream.call.startBatch(batch, function(){}); } @@ -239,6 +250,10 @@ function ServerWritableStream(call, serialize) { function _write(chunk, encoding, callback) { /* jshint validthis: true */ var batch = {}; + if (!this.call.metadataSent) { + batch[grpc.opType.SEND_INITIAL_METADATA] = {}; + this.call.metadataSent = true; + } batch[grpc.opType.SEND_MESSAGE] = this.serialize(chunk); this.call.startBatch(batch, function(err, value) { if (err) { @@ -251,6 +266,23 @@ function _write(chunk, encoding, callback) { ServerWritableStream.prototype._write = _write; +function sendMetadata(responseMetadata) { + /* jshint validthis: true */ + if (!this.call.metadataSent) { + this.call.metadataSent = true; + var batch = []; + batch[grpc.opType.SEND_INITIAL_METADATA] = responseMetadata; + this.call.startBatch(batch, function(err) { + if (err) { + this.emit('error', err); + return; + } + }); + } +} + +ServerWritableStream.prototype.sendMetadata = sendMetadata; + util.inherits(ServerReadableStream, Readable); /** @@ -339,6 +371,20 @@ function ServerDuplexStream(call, serialize, deserialize) { 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 @@ -348,12 +394,23 @@ ServerDuplexStream.prototype._write = _write; */ function handleUnary(call, handler, metadata) { var emitter = new EventEmitter(); + emitter.sendMetadata = function(responseMetadata) { + if (!call.metadataSent) { + call.metadataSent = true; + var batch = {}; + batch[grpc.opType.SEND_INITIAL_METADATA] = responseMetadata; + call.startBatch(batch, function() {}); + } + }; + emitter.getPeer = function() { + return call.getPeer(); + }; emitter.on('error', function(error) { handleError(call, error); }); + emitter.metadata = metadata; waitForCancel(call, emitter); var batch = {}; - batch[grpc.opType.SEND_INITIAL_METADATA] = metadata; batch[grpc.opType.RECV_MESSAGE] = true; call.startBatch(batch, function(err, result) { if (err) { @@ -392,8 +449,8 @@ function handleUnary(call, handler, metadata) { function handleServerStreaming(call, handler, metadata) { var stream = new ServerWritableStream(call, handler.serialize); waitForCancel(call, stream); + stream.metadata = metadata; var batch = {}; - batch[grpc.opType.SEND_INITIAL_METADATA] = metadata; batch[grpc.opType.RECV_MESSAGE] = true; call.startBatch(batch, function(err, result) { if (err) { @@ -419,13 +476,19 @@ function handleServerStreaming(call, handler, metadata) { */ function handleClientStreaming(call, handler, metadata) { var stream = new ServerReadableStream(call, handler.deserialize); + stream.sendMetadata = function(responseMetadata) { + if (!call.metadataSent) { + call.metadataSent = true; + var batch = {}; + batch[grpc.opType.SEND_INITIAL_METADATA] = responseMetadata; + call.startBatch(batch, function() {}); + } + }; stream.on('error', function(error) { handleError(call, error); }); waitForCancel(call, stream); - var metadata_batch = {}; - metadata_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata; - call.startBatch(metadata_batch, function() {}); + stream.metadata = metadata; handler.func(stream, function(err, value, trailer) { stream.terminate(); if (err) { @@ -449,9 +512,7 @@ function handleBidiStreaming(call, handler, metadata) { var stream = new ServerDuplexStream(call, handler.serialize, handler.deserialize); waitForCancel(call, stream); - var metadata_batch = {}; - metadata_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata; - call.startBatch(metadata_batch, function() {}); + stream.metadata = metadata; handler.func(stream); } @@ -466,29 +527,28 @@ var streamHandlers = { * Constructs a server object that stores request handlers and delegates * incoming requests to those handlers * @constructor - * @param {function(string, Object<string, Array<Buffer>>): - Object<string, Array<Buffer|string>>=} getMetadata Callback that gets - * metatada for a given method * @param {Object=} options Options that should be passed to the internal server * implementation */ -function Server(getMetadata, options) { +function Server(options) { this.handlers = {}; var handlers = this.handlers; var server = new grpc.Server(options); this._server = server; + this.started = false; /** * Start the server and begin handling requests * @this Server */ - this.listen = function() { + this.start = function() { + if (this.started) { + throw new Error('Server is already running'); + } + this.started = true; console.log('Server starting'); _.each(handlers, function(handler, handler_name) { console.log('Serving', handler_name); }); - if (this.started) { - throw 'Server is already running'; - } server.start(); /** * Handles the SERVER_RPC_NEW event. If there is a handler associated with @@ -500,7 +560,7 @@ function Server(getMetadata, 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; @@ -523,11 +583,7 @@ function Server(getMetadata, options) { call.startBatch(batch, function() {}); return; } - var response_metadata = {}; - if (getMetadata) { - response_metadata = getMetadata(method, metadata); - } - streamHandlers[handler.type](call, handler, response_metadata); + streamHandlers[handler.type](call, handler, metadata); } server.requestCall(handleNewCall); }; @@ -565,6 +621,47 @@ Server.prototype.register = function(name, handler, serialize, deserialize, return true; }; +Server.prototype.addService = function(service, implementation) { + if (this.started) { + throw new Error('Can\'t add a service to a started server.'); + } + var self = this; + _.each(service, function(attrs, name) { + var method_type; + if (attrs.requestStream) { + if (attrs.responseStream) { + method_type = 'bidi'; + } else { + method_type = 'client_stream'; + } + } else { + if (attrs.responseStream) { + method_type = 'server_stream'; + } else { + method_type = 'unary'; + } + } + if (implementation[name] === undefined) { + throw new Error('Method handler for ' + attrs.path + + ' not provided.'); + } + var serialize = attrs.responseSerialize; + var deserialize = attrs.requestDeserialize; + var register_success = self.register(attrs.path, + _.bind(implementation[name], + implementation), + serialize, deserialize, method_type); + if (!register_success) { + throw new Error('Method handler for ' + attrs.path + + ' already provided.'); + } + }); +}; + +Server.prototype.addProtoService = function(service, implementation) { + this.addService(common.getProtobufServiceAttrs(service), implementation); +}; + /** * Binds the server to the given port, with SSL enabled if creds is given * @param {string} port The port that the server should bind on, in the format @@ -573,6 +670,9 @@ Server.prototype.register = function(name, handler, serialize, deserialize, * nothing for an insecure port */ 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 { @@ -581,131 +681,6 @@ Server.prototype.bind = function(port, creds) { }; /** - * Create a constructor for servers with services defined by service_attr_map. - * That is an object that maps (namespaced) service names to objects that in - * turn map method names to objects with the following keys: - * path: The path on the server for accessing the method. For example, for - * protocol buffers, we use "/service_name/method_name" - * requestStream: bool indicating whether the client sends a stream - * resonseStream: bool indicating whether the server sends a stream - * requestDeserialize: function to deserialize request objects - * responseSerialize: function to serialize response objects - * @param {Object} service_attr_map An object mapping service names to method - * attribute map objects - * @return {function(Object, function, Object=)} New server constructor - */ -function makeServerConstructor(service_attr_map) { - /** - * Create a server with the given handlers for all of the methods. - * @constructor - * @param {Object} service_handlers Map from service names to map from method - * names to handlers - * @param {function(string, Object<string, Array<Buffer>>): - Object<string, Array<Buffer|string>>=} getMetadata Callback that - * gets metatada for a given method - * @param {Object=} options Options to pass to the underlying server - */ - function SurfaceServer(service_handlers, getMetadata, options) { - var server = new Server(getMetadata, options); - this.inner_server = server; - _.each(service_attr_map, function(service_attrs, service_name) { - if (service_handlers[service_name] === undefined) { - throw new Error('Handlers for service ' + - service_name + ' not provided.'); - } - _.each(service_attrs, function(attrs, name) { - var method_type; - if (attrs.requestStream) { - if (attrs.responseStream) { - method_type = 'bidi'; - } else { - method_type = 'client_stream'; - } - } else { - if (attrs.responseStream) { - method_type = 'server_stream'; - } else { - method_type = 'unary'; - } - } - if (service_handlers[service_name][name] === undefined) { - throw new Error('Method handler for ' + attrs.path + - ' not provided.'); - } - var serialize = attrs.responseSerialize; - var deserialize = attrs.requestDeserialize; - server.register(attrs.path, _.bind(service_handlers[service_name][name], - service_handlers[service_name]), - serialize, deserialize, method_type); - }); - }, this); - } - - /** - * Binds the server to the given port, with SSL enabled if creds is supplied - * @param {string} port The port that the server should bind on, in the format - * "address:port" - * @param {boolean=} creds Credentials to use for SSL - * @return {SurfaceServer} this - */ - SurfaceServer.prototype.bind = function(port, creds) { - return this.inner_server.bind(port, creds); - }; - - /** - * Starts the server listening on any bound ports - * @return {SurfaceServer} this - */ - SurfaceServer.prototype.listen = function() { - this.inner_server.listen(); - return this; - }; - - /** - * Shuts the server down; tells it to stop listening for new requests and to - * kill old requests. - */ - SurfaceServer.prototype.shutdown = function() { - this.inner_server.shutdown(); - }; - - return SurfaceServer; -} - -/** - * Create a constructor for servers that serve the given services. - * @param {Array<ProtoBuf.Reflect.Service>} services The services that the - * servers will serve - * @return {function(Object, function, Object=)} New server constructor - */ -function makeProtobufServerConstructor(services) { - var qual_names = []; - var service_attr_map = {}; - _.each(services, function(service) { - var service_name = common.fullyQualifiedName(service); - _.each(service.children, function(method) { - var name = common.fullyQualifiedName(method); - if (_.indexOf(qual_names, name) !== -1) { - throw new Error('Method ' + name + ' exposed by more than one service'); - } - qual_names.push(name); - }); - var method_attrs = common.getProtobufServiceAttrs(service); - if (!service_attr_map.hasOwnProperty(service_name)) { - service_attr_map[service_name] = {}; - } - service_attr_map[service_name] = _.extend(service_attr_map[service_name], - method_attrs); - }); - return makeServerConstructor(service_attr_map); -} - -/** - * See documentation for makeServerConstructor - */ -exports.makeServerConstructor = makeServerConstructor; - -/** - * See documentation for makeProtobufServerConstructor + * See documentation for Server */ -exports.makeProtobufServerConstructor = makeProtobufServerConstructor; +exports.Server = Server; diff --git a/src/node/test/call_test.js b/src/node/test/call_test.js index 98158ffff3..942c31ac68 100644 --- a/src/node/test/call_test.js +++ b/src/node/test/call_test.js @@ -132,7 +132,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 +147,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 +184,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/src/node/test/channel_test.js b/src/node/test/channel_test.js index 33200c99ee..3e61d3bbc6 100644 --- a/src/node/test/channel_test.js +++ b/src/node/test/channel_test.js @@ -87,4 +87,10 @@ describe('channel', function() { }); }); }); + describe('getTarget', function() { + it('should return a string', function() { + var channel = new grpc.Channel('localhost', {}); + assert.strictEqual(typeof channel.getTarget(), 'string'); + }); + }); }); diff --git a/src/node/test/end_to_end_test.js b/src/node/test/end_to_end_test.js index 667852f382..5d3baf823d 100644 --- a/src/node/test/end_to_end_test.js +++ b/src/node/test/end_to_end_test.js @@ -85,37 +85,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(); }); @@ -131,7 +131,7 @@ describe('end-to-end', function() { 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 +139,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 +158,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(); }); @@ -196,19 +196,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 +217,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(); }); @@ -251,9 +251,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 +262,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 +275,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 +284,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 +294,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/src/node/test/health_test.js b/src/node/test/health_test.js index 4d1a5082e0..bb700cc46c 100644 --- a/src/node/test/health_test.js +++ b/src/node/test/health_test.js @@ -49,14 +49,13 @@ describe('Health Checking', function() { 'grpc.test.TestService': 'SERVING' } }; - var HealthServer = grpc.buildServer([health.service]); - var healthServer = new HealthServer({ - 'grpc.health.v1alpha.Health': new health.Implementation(statusMap) - }); + var healthServer = new grpc.Server(); + healthServer.addProtoService(health.service, + new health.Implementation(statusMap)); var healthClient; before(function() { var port_num = healthServer.bind('0.0.0.0:0'); - healthServer.listen(); + healthServer.start(); healthClient = new health.Client('localhost:' + port_num); }); after(function() { diff --git a/src/node/test/interop_sanity_test.js b/src/node/test/interop_sanity_test.js index fcd8eb6403..0a5eb29c0c 100644 --- a/src/node/test/interop_sanity_test.js +++ b/src/node/test/interop_sanity_test.js @@ -46,7 +46,7 @@ describe('Interop tests', function() { before(function(done) { var server_obj = interop_server.getServer(0, true); server = server_obj.server; - server.listen(); + server.start(); port = 'localhost:' + server_obj.port; done(); }); diff --git a/src/node/test/math_client_test.js b/src/node/test/math_client_test.js index 3461922e66..f2751857ff 100644 --- a/src/node/test/math_client_test.js +++ b/src/node/test/math_client_test.js @@ -52,7 +52,7 @@ var server = require('../examples/math_server.js'); describe('Math client', function() { before(function(done) { var port_num = server.bind('0.0.0.0:0'); - server.listen(); + server.start(); math_client = new math.Math('localhost:' + port_num); done(); }); diff --git a/src/node/test/server_test.js b/src/node/test/server_test.js index 7cb34fa0cb..9c7bb465aa 100644 --- a/src/node/test/server_test.js +++ b/src/node/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() { @@ -67,9 +69,13 @@ describe('server', function() { before(function() { server = new grpc.Server(); }); - it('should bind to an unused port with fake credentials', function() { + it('should bind to an unused port with ssl credentials', function() { var port; - var creds = grpc.ServerCredentials.createFake(); + 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.addSecureHttp2Port('0.0.0.0:0', creds); }); diff --git a/src/node/test/surface_test.js b/src/node/test/surface_test.js index 125957277f..9005cbd505 100644 --- a/src/node/test/surface_test.js +++ b/src/node/test/surface_test.js @@ -69,34 +69,45 @@ describe('File loader', function() { }); }); }); -describe('Surface server constructor', function() { - it('Should fail with conflicting method names', function() { - assert.throws(function() { - grpc.buildServer([mathService, mathService]); - }); +describe('Server.prototype.addProtoService', function() { + var server; + var dummyImpls = { + 'div': function() {}, + 'divMany': function() {}, + 'fib': function() {}, + 'sum': function() {} + }; + beforeEach(function() { + server = new grpc.Server(); + }); + afterEach(function() { + server.shutdown(); }); it('Should succeed with a single service', function() { assert.doesNotThrow(function() { - grpc.buildServer([mathService]); + server.addProtoService(mathService, dummyImpls); + }); + }); + it('Should fail with conflicting method names', function() { + server.addProtoService(mathService, dummyImpls); + assert.throws(function() { + server.addProtoService(mathService, dummyImpls); }); }); it('Should fail with missing handlers', function() { - var Server = grpc.buildServer([mathService]); assert.throws(function() { - new Server({ - 'math.Math': { - 'div': function() {}, - 'divMany': function() {}, - 'fib': function() {} - } + server.addProtoService(mathService, { + 'div': function() {}, + 'divMany': function() {}, + 'fib': function() {} }); }, /math.Math.Sum/); }); - it('Should fail with no handlers for the service', function() { - var Server = grpc.buildServer([mathService]); + it('Should fail if the server has been started', function() { + server.start(); assert.throws(function() { - new Server({}); - }, /math.Math/); + server.addProtoService(mathService, dummyImpls); + }); }); }); describe('Echo service', function() { @@ -105,18 +116,16 @@ describe('Echo service', function() { before(function() { var test_proto = ProtoBuf.loadProtoFile(__dirname + '/echo_service.proto'); var echo_service = test_proto.lookup('EchoService'); - var Server = grpc.buildServer([echo_service]); - server = new Server({ - 'EchoService': { - echo: function(call, callback) { - callback(null, call.request); - } + server = new grpc.Server(); + server.addProtoService(echo_service, { + echo: function(call, callback) { + callback(null, call.request); } }); var port = server.bind('localhost:0'); var Client = surface_client.makeProtobufClientConstructor(echo_service); client = new Client('localhost:' + port); - server.listen(); + server.start(); }); after(function() { server.shutdown(); @@ -151,18 +160,14 @@ describe('Generic client and server', function() { var client; var server; before(function() { - var Server = grpc.makeGenericServerConstructor({ - string: string_service_attrs - }); - server = new Server({ - string: { - capitalize: function(call, callback) { - callback(null, _.capitalize(call.request)); - } + server = new grpc.Server(); + server.addService(string_service_attrs, { + capitalize: function(call, callback) { + callback(null, _.capitalize(call.request)); } }); var port = server.bind('localhost:0'); - server.listen(); + server.start(); var Client = grpc.makeGenericClientConstructor(string_service_attrs); client = new Client('localhost:' + port); }); @@ -178,6 +183,92 @@ describe('Generic client and server', function() { }); }); }); +describe('Echo metadata', function() { + var client; + var server; + before(function() { + var test_proto = ProtoBuf.loadProtoFile(__dirname + '/test_service.proto'); + var test_service = test_proto.lookup('TestService'); + server = new grpc.Server(); + server.addProtoService(test_service, { + unary: function(call, cb) { + call.sendMetadata(call.metadata); + cb(null, {}); + }, + clientStream: function(stream, cb){ + stream.on('data', function(data) {}); + stream.on('end', function() { + stream.sendMetadata(stream.metadata); + cb(null, {}); + }); + }, + serverStream: function(stream) { + stream.sendMetadata(stream.metadata); + stream.end(); + }, + bidiStream: function(stream) { + stream.on('data', function(data) {}); + stream.on('end', function() { + stream.sendMetadata(stream.metadata); + stream.end(); + }); + } + }); + var port = server.bind('localhost:0'); + var Client = surface_client.makeProtobufClientConstructor(test_service); + client = new Client('localhost:' + port); + server.start(); + }); + after(function() { + server.shutdown(); + }); + it('with unary call', function(done) { + var call = client.unary({}, function(err, data) { + assert.ifError(err); + }, {key: ['value']}); + call.on('metadata', function(metadata) { + assert.deepEqual(metadata.key, ['value']); + done(); + }); + }); + it('with client stream call', function(done) { + var call = client.clientStream(function(err, data) { + assert.ifError(err); + }, {key: ['value']}); + call.on('metadata', function(metadata) { + assert.deepEqual(metadata.key, ['value']); + done(); + }); + call.end(); + }); + it('with server stream call', function(done) { + var call = client.serverStream({}, {key: ['value']}); + call.on('data', function() {}); + call.on('metadata', function(metadata) { + assert.deepEqual(metadata.key, ['value']); + done(); + }); + }); + it('with bidi stream call', function(done) { + var call = client.bidiStream({key: ['value']}); + call.on('data', function() {}); + call.on('metadata', function(metadata) { + assert.deepEqual(metadata.key, ['value']); + done(); + }); + 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 client; var server; @@ -185,76 +276,77 @@ describe('Other conditions', function() { before(function() { var test_proto = ProtoBuf.loadProtoFile(__dirname + '/test_service.proto'); var test_service = test_proto.lookup('TestService'); - var Server = grpc.buildServer([test_service]); - server = new Server({ - TestService: { - unary: function(call, cb) { - var req = call.request; - if (req.error) { - cb(new Error('Requested error'), null, {metadata: ['yes']}); + server = new grpc.Server(); + server.addProtoService(test_service, { + unary: function(call, cb) { + var req = call.request; + if (req.error) { + cb(new Error('Requested error'), null, {trailer_present: ['yes']}); + } else { + cb(null, {count: 1}, {trailer_present: ['yes']}); + } + }, + clientStream: function(stream, cb){ + var count = 0; + var errored; + stream.on('data', function(data) { + if (data.error) { + errored = true; + cb(new Error('Requested error'), null, {trailer_present: ['yes']}); } else { - cb(null, {count: 1}, {metadata: ['yes']}); + count += 1; } - }, - clientStream: function(stream, cb){ - var count = 0; - var errored; - stream.on('data', function(data) { - if (data.error) { - errored = true; - cb(new Error('Requested error'), null, {metadata: ['yes']}); - } else { - count += 1; - } - }); - stream.on('end', function() { - if (!errored) { - cb(null, {count: count}, {metadata: ['yes']}); - } - }); - }, - serverStream: function(stream) { - var req = stream.request; - if (req.error) { + }); + stream.on('end', function() { + if (!errored) { + cb(null, {count: count}, {trailer_present: ['yes']}); + } + }); + }, + serverStream: function(stream) { + var req = stream.request; + if (req.error) { + var err = new Error('Requested error'); + err.metadata = {trailer_present: ['yes']}; + stream.emit('error', err); + } else { + for (var i = 0; i < 5; i++) { + stream.write({count: i}); + } + stream.end({trailer_present: ['yes']}); + } + }, + bidiStream: function(stream) { + var count = 0; + stream.on('data', function(data) { + if (data.error) { var err = new Error('Requested error'); - err.metadata = {metadata: ['yes']}; + err.metadata = { + trailer_present: ['yes'], + count: ['' + count] + }; stream.emit('error', err); } else { - for (var i = 0; i < 5; i++) { - stream.write({count: i}); - } - stream.end({metadata: ['yes']}); + stream.write({count: count}); + count += 1; } - }, - bidiStream: function(stream) { - var count = 0; - stream.on('data', function(data) { - if (data.error) { - var err = new Error('Requested error'); - err.metadata = { - metadata: ['yes'], - count: ['' + count] - }; - stream.emit('error', err); - } else { - stream.write({count: count}); - count += 1; - } - }); - stream.on('end', function() { - stream.end({metadata: ['yes']}); - }); - } + }); + stream.on('end', function() { + stream.end({trailer_present: ['yes']}); + }); } }); port = server.bind('localhost:0'); var Client = surface_client.makeProtobufClientConstructor(test_service); client = new Client('localhost:' + port); - server.listen(); + 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]); @@ -340,7 +432,7 @@ describe('Other conditions', function() { assert.ifError(err); }); call.on('status', function(status) { - assert.deepEqual(status.metadata.metadata, ['yes']); + assert.deepEqual(status.metadata.trailer_present, ['yes']); done(); }); }); @@ -349,7 +441,7 @@ describe('Other conditions', function() { assert(err); }); call.on('status', function(status) { - assert.deepEqual(status.metadata.metadata, ['yes']); + assert.deepEqual(status.metadata.trailer_present, ['yes']); done(); }); }); @@ -361,7 +453,7 @@ describe('Other conditions', function() { call.write({error: false}); call.end(); call.on('status', function(status) { - assert.deepEqual(status.metadata.metadata, ['yes']); + assert.deepEqual(status.metadata.trailer_present, ['yes']); done(); }); }); @@ -373,7 +465,7 @@ describe('Other conditions', function() { call.write({error: true}); call.end(); call.on('status', function(status) { - assert.deepEqual(status.metadata.metadata, ['yes']); + assert.deepEqual(status.metadata.trailer_present, ['yes']); done(); }); }); @@ -382,7 +474,7 @@ describe('Other conditions', function() { call.on('data', function(){}); call.on('status', function(status) { assert.strictEqual(status.code, grpc.status.OK); - assert.deepEqual(status.metadata.metadata, ['yes']); + assert.deepEqual(status.metadata.trailer_present, ['yes']); done(); }); }); @@ -390,7 +482,7 @@ describe('Other conditions', function() { var call = client.serverStream({error: true}); call.on('data', function(){}); call.on('error', function(error) { - assert.deepEqual(error.metadata.metadata, ['yes']); + assert.deepEqual(error.metadata.trailer_present, ['yes']); done(); }); }); @@ -402,7 +494,7 @@ describe('Other conditions', function() { call.on('data', function(){}); call.on('status', function(status) { assert.strictEqual(status.code, grpc.status.OK); - assert.deepEqual(status.metadata.metadata, ['yes']); + assert.deepEqual(status.metadata.trailer_present, ['yes']); done(); }); }); @@ -413,7 +505,7 @@ describe('Other conditions', function() { call.end(); call.on('data', function(){}); call.on('error', function(error) { - assert.deepEqual(error.metadata.metadata, ['yes']); + assert.deepEqual(error.metadata.trailer_present, ['yes']); done(); }); }); @@ -460,23 +552,59 @@ 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('Cancelling surface client', function() { var client; var server; before(function() { - var Server = grpc.buildServer([mathService]); - server = new Server({ - 'math.Math': { - 'div': function(stream) {}, - 'divMany': function(stream) {}, - 'fib': function(stream) {}, - 'sum': function(stream) {} - } + server = new grpc.Server(); + server.addProtoService(mathService, { + 'div': function(stream) {}, + 'divMany': function(stream) {}, + 'fib': function(stream) {}, + 'sum': function(stream) {} }); var port = server.bind('localhost:0'); var Client = surface_client.makeProtobufClientConstructor(mathService); client = new Client('localhost:' + port); + server.start(); }); after(function() { server.shutdown(); |