archive-gh-me/grpc-node
1
0
Fork 0
mirror of https://github.com/grpc/grpc-node.git synced 2025-12-08 18:23:54 +00:00
Code Issues Projects Releases Wiki Activity

Refactor some code and document most of the API

Browse Source
This commit is contained in:
murgatroid99 2017-05-23 10:28:47 -07:00
parent 27836dadee
commit 5309a59a24
12 changed files with 660 additions and 455 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
README.md
View File

@ -2,9 +2,9 @@
# Node.js gRPC Library
## PREREQUISITES
- `node`: This requires `node` to be installed, version `0.12` or above. If you instead have the `nodejs` executable on Debian, you should install the [`nodejs-legacy`](https://packages.debian.org/sid/nodejs-legacy) package.
- `node`: This requires `node` to be installed, version `4.0` or above. If you instead have the `nodejs` executable on Debian, you should install the [`nodejs-legacy`](https://packages.debian.org/sid/nodejs-legacy) package.
- **Note:** If you installed `node` via a package manager and the version is still less than `0.12`, try directly installing it from [nodejs.org](https://nodejs.org).
- **Note:** If you installed `node` via a package manager and the version is still less than `4.0`, try directly installing it from [nodejs.org](https://nodejs.org).
## INSTALLATION
@ -16,7 +16,7 @@ npm install grpc
## BUILD FROM SOURCE
1. Clone [the grpc Git Repository](https://github.com/grpc/grpc).
2. Run `npm install` from the repository root.
2. Run `npm install --build-from-source` from the repository root.
- **Note:** On Windows, this might fail due to [nodejs issue #4932](https://github.com/nodejs/node/issues/4932) in which case, you will see something like the following in `npm install`'s output (towards the very beginning):
@ -34,61 +34,3 @@ npm install grpc
## TESTING
To run the test suite, simply run `npm test` in the install location.
## API
This library internally uses [ProtoBuf.js](https://github.com/dcodeIO/ProtoBuf.js), and some structures it exports match those exported by that library.
If you require this module, you will get an object with the following members
```javascript
function load(filename)
```
Takes a filename of a [Protocol Buffer](https://developers.google.com/protocol-buffers/) file, and returns an object representing the structure of the protocol buffer in the following way:
- Namespaces become maps from the names of their direct members to those member objects
- Service definitions become client constructors for clients for that service. They also have a `service` member that can be used for constructing servers.
- Message definitions become Message constructors like those that ProtoBuf.js would create
- Enum definitions become Enum objects like those that ProtoBuf.js would create
- Anything else becomes the relevant reflection object that ProtoBuf.js would create
```javascript
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 Server([serverOptions])
```
Constructs a server to which service/implementation pairs can be added.
```javascript
status
```
An object mapping status names to status code numbers.
```javascript
callError
```
An object mapping call error names to codes. This is primarily useful for tracking down certain kinds of internal errors.
```javascript
Credentials
```
An object with factory methods for creating credential objects for clients.
```javascript
ServerCredentials
```
An object with factory methods for creating credential objects for servers.

146
index.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2015, Google Inc.
* All rights reserved.
*
@ -31,10 +31,6 @@
*
*/
/**
* @module
*/
'use strict';
var path = require('path');
@ -64,26 +60,30 @@ var constants = require('./src/constants.js');
grpc.setDefaultRootsPem(fs.readFileSync(SSL_ROOTS_PATH, 'ascii'));
/**
* Load a ProtoBuf.js object as a gRPC object. The options object can provide
* the following options:
* - binaryAsBase64: deserialize bytes values as base64 strings instead of
* Buffers. Defaults to false
* - longsAsStrings: deserialize long values as strings instead of objects.
* Defaults to true
* - enumsAsStrings: deserialize enum values as strings instead of numbers.
* Defaults to true
* - deprecatedArgumentOrder: Use the beta method argument order for client
* methods, with optional arguments after the callback. Defaults to false.
* This option is only a temporary stopgap measure to smooth an API breakage.
* It is deprecated, and new code should not use it.
* - protobufjsVersion: Available values are 5, 6, and 'detect'. 5 and 6
* respectively indicate that an object from the corresponding version of
* ProtoBuf.js is provided in the value argument. If the option is 'detect',
* gRPC will guess what the version is based on the structure of the value.
* Defaults to 'detect'.
* @namespace grpc
*/
/**
* Load a ProtoBuf.js object as a gRPC object.
* @memberof grpc
* @alias grpc.loadObject
* @param {Object} value The ProtoBuf.js reflection object to load
* @param {Object=} options Options to apply to the loaded file
* @return {Object<string, *>} The resulting gRPC object
* @param {bool=} [options.binaryAsBase64=false] deserialize bytes values as
* base64 strings instead of Buffers
* @param {bool=} [options.longsAsStrings=true] deserialize long values as
* strings instead of objects
* @param {bool=} [options.enumsAsStrings=true] deserialize enum values as
* strings instead of numbers. Only works with Protobuf.js 6 values.
* @param {bool=} [options.deprecatedArgumentOrder=false] use the beta method
* argument order for client methods, with optional arguments after the
* callback. This option is only a temporary stopgap measure to smooth an
* API breakage. It is deprecated, and new code should not use it.
* @param {(number|string)=} [options.protobufjsVersion='detect'] 5 and 6
* respectively indicate that an object from the corresponding version of
* Protobuf.js is provided in the value argument. If the option is 'detect',
* gRPC wll guess what the version is based on the structure of the value.
* @return {Object<string, *>} The resulting gRPC object.
*/
exports.loadObject = function loadObject(value, options) {
options = _.defaults(options, common.defaultGrpcOptions);
@ -131,24 +131,23 @@ function applyProtoRoot(filename, root) {
}
/**
* Load a gRPC object from a .proto file. The options object can provide the
* following options:
* - convertFieldsToCamelCase: Load this file with field names in camel case
* instead of their original case
* - binaryAsBase64: deserialize bytes values as base64 strings instead of
* Buffers. Defaults to false
* - longsAsStrings: deserialize long values as strings instead of objects.
* Defaults to true
* - enumsAsStrings: deserialize enum values as strings instead of numbers.
* Defaults to true
* - deprecatedArgumentOrder: Use the beta method argument order for client
* methods, with optional arguments after the callback. Defaults to false.
* This option is only a temporary stopgap measure to smooth an API breakage.
* It is deprecated, and new code should not use it.
* Load a gRPC object from a .proto file.
* @memberof grpc
* @alias grpc.load
* @param {string|{root: string, file: string}} filename The file to load
* @param {string=} format The file format to expect. Must be either 'proto' or
* 'json'. Defaults to 'proto'
* @param {Object=} options Options to apply to the loaded file
* @param {bool=} [options.convertFieldsToCamelCase=false] Load this file with
* field names in camel case instead of their original case
* @param {bool=} [options.binaryAsBase64=false] deserialize bytes values as
* base64 strings instead of Buffers
* @param {bool=} [options.longsAsStrings=true] deserialize long values as
* strings instead of objects
* @param {bool=} [options.deprecatedArgumentOrder=false] use the beta method
* argument order for client methods, with optional arguments after the
* callback. This option is only a temporary stopgap measure to smooth an
* API breakage. It is deprecated, and new code should not use it.
* @return {Object<string, *>} The resulting gRPC object
*/
exports.load = function load(filename, format, options) {
@ -175,6 +174,8 @@ var log_template = _.template(
* called. Note: the output format here is intended to be informational, and
* is not guaranteed to stay the same in the future.
* Logs will be directed to logger.error.
* @memberof grpc
* @alias grpc.setLogger
* @param {Console} logger A Console-like object.
*/
exports.setLogger = function setLogger(logger) {
@ -194,6 +195,8 @@ exports.setLogger = function setLogger(logger) {
/**
* Sets the logger verbosity for gRPC module logging. The options are members
* of the grpc.logVerbosity map.
* @memberof grpc
* @alias grpc.setLogVerbosity
* @param {Number} verbosity The minimum severity to log
*/
exports.setLogVerbosity = function setLogVerbosity(verbosity) {
@ -201,71 +204,70 @@ exports.setLogVerbosity = function setLogVerbosity(verbosity) {
grpc.setLogVerbosity(verbosity);
};
/**
* @see module:src/server.Server
*/
exports.Server = server.Server;
/**
* @see module:src/metadata
*/
exports.Metadata = Metadata;
/**
* Status name to code number mapping
*/
exports.status = constants.status;
/**
* Propagate flag name to number mapping
*/
exports.propagate = constants.propagate;
/**
* Call error name to code number mapping
*/
exports.callError = constants.callError;
/**
* Write flag name to code number mapping
*/
exports.writeFlags = constants.writeFlags;
/**
* Log verbosity setting name to code number mapping
*/
exports.logVerbosity = constants.logVerbosity;
/**
* Credentials factories
*/
exports.credentials = require('./src/credentials.js');
/**
* ServerCredentials factories
* @constructor ServerCredentials
* @memberof grpc
*/
exports.ServerCredentials = grpc.ServerCredentials;
/**
* @see module:src/client.makeClientConstructor
* Create insecure server credentials
* @name grpc.ServerCredentials.createInsecure
* @kind function
* @return grpc.ServerCredentials
*/
/**
* A private key and certificate pair
* @typedef {Object} grpc.ServerCredentials~keyCertPair
* @property {Buffer} privateKey The server's private key
* @property {Buffer} certChain The server's certificate chain
*/
/**
* Create SSL server credentials
* @name grpc.ServerCredentials.createInsecure
* @kind function
* @param {?Buffer} rootCerts Root CA certificates for validating client
* certificates
* @param {Array<grpc.ServerCredentials~keyCertPair>} keyCertPairs A list of
* private key and certificate chain pairs to be used for authenticating
* the server
* @param {boolean} [checkClientCertificate=false] Indicates that the server
* should request and verify the client's certificates
* @return grpc.ServerCredentials
*/
exports.makeGenericClientConstructor = client.makeClientConstructor;
/**
* @see module:src/client.getClientChannel
*/
exports.getClientChannel = client.getClientChannel;
/**
* @see module:src/client.waitForClientReady
*/
exports.waitForClientReady = client.waitForClientReady;
/**
* @memberof grpc
* @alias grpc.closeClient
* @param {grpc.Client} client_obj The client to close
*/
exports.closeClient = function closeClient(client_obj) {
client.Client.prototype.close.apply(client_obj);
};
/**
* @see module:src/client.Client
*/
exports.Client = client.Client;

256
src/client.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2015, Google Inc.
* All rights reserved.
*
@ -43,8 +43,6 @@
* var Client = proto_obj.package.subpackage.ServiceName;
* var client = new Client(server_address, client_credentials);
* var call = client.unaryMethod(arguments, callback);
*
* @module
*/
'use strict';
@ -70,13 +68,26 @@ var Duplex = stream.Duplex;
var util = require('util');
var version = require('../../../package.json').version;
/**
* Initial response metadata sent by the server when it starts processing the
* call
* @event grpc~ClientUnaryCall#metadata
* @type {grpc.Metadata}
*/
/**
* Status of the call when it has completed.
* @event grpc~ClientUnaryCall#status
* @type grpc~StatusObject
*/
util.inherits(ClientUnaryCall, EventEmitter);
/**
* An EventEmitter. Used for unary calls
* @constructor
* An EventEmitter. Used for unary calls.
* @constructor grpc~ClientUnaryCall
* @extends external:EventEmitter
* @param {grpc.Call} call The call object associated with the request
* @param {grpc.internal~Call} call The call object associated with the request
*/
function ClientUnaryCall(call) {
EventEmitter.call(this);
@ -88,14 +99,16 @@ util.inherits(ClientWritableStream, Writable);
/**
* A stream that the client can write to. Used for calls that are streaming from
* the client side.
* @constructor
* @constructor grpc~ClientWritableStream
* @extends external:Writable
* @borrows module:src/client~ClientUnaryCall#cancel as
* module:src/client~ClientWritableStream#cancel
* @borrows module:src/client~ClientUnaryCall#getPeer as
* module:src/client~ClientWritableStream#getPeer
* @param {grpc.Call} call The call object to send data with
* @param {module:src/common~serialize=} [serialize=identity] Serialization
* @borrows grpc~ClientUnaryCall#cancel as grpc~ClientWritableStream#cancel
* @borrows grpc~ClientUnaryCall#getPeer as grpc~ClientWritableStream#getPeer
* @borrows grpc~ClientUnaryCall#event:metadata as
* grpc~ClientWritableStream#metadata
* @borrows grpc~ClientUnaryCall#event:status as
* grpc~ClientWritableStream#status
* @param {grpc.internal~Call} call The call object to send data with
* @param {grpc~serialize=} [serialize=identity] Serialization
* function for writes.
*/
function ClientWritableStream(call, serialize) {
@ -109,12 +122,25 @@ function ClientWritableStream(call, serialize) {
});
}
/**
* Write a message to the request stream. If serializing the argument fails,
* the call will be cancelled and the stream will end with an error.
* @name grpc~ClientWritableStream#write
* @kind function
* @override
* @param {*} message The message to write. Must be a valid argument to the
* serialize function of the corresponding method
* @param {grpc.writeFlags} flags Flags to modify how the message is written
* @param {Function} callback Callback for when this chunk of data is flushed
* @return {boolean} As defined for [Writable]{@link external:Writable}
*/
/**
* 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
* @private
* @param {*} chunk The chunk to write
* @param {grpc.writeFlags} encoding Used to pass write flags
* @param {function(Error=)} callback Called when the write is complete
*/
function _write(chunk, encoding, callback) {
@ -155,14 +181,16 @@ util.inherits(ClientReadableStream, Readable);
/**
* A stream that the client can read from. Used for calls that are streaming
* from the server side.
* @constructor
* @constructor grpc~ClientReadableStream
* @extends external:Readable
* @borrows module:src/client~ClientUnaryCall#cancel as
* module:src/client~ClientReadableStream#cancel
* @borrows module:src/client~ClientUnaryCall#getPeer as
* module:src/client~ClientReadableStream#getPeer
* @param {grpc.Call} call The call object to read data with
* @param {module:src/common~deserialize=} [deserialize=identity]
* @borrows grpc~ClientUnaryCall#cancel as grpc~ClientReadableStream#cancel
* @borrows grpc~ClientUnaryCall#getPeer as grpc~ClientReadableStream#getPeer
* @borrows grpc~ClientUnaryCall#event:metadata as
* grpc~ClientReadableStream#metadata
* @borrows grpc~ClientUnaryCall#event:status as
* grpc~ClientReadableStream#status
* @param {grpc.internal~Call} call The call object to read data with
* @param {grpc~deserialize=} [deserialize=identity]
* Deserialization function for reads
*/
function ClientReadableStream(call, deserialize) {
@ -183,7 +211,7 @@ function ClientReadableStream(call, deserialize) {
* parameter indicates that the call should end with that status. status
* defaults to OK if not provided.
* @param {Object!} status The status that the call should end with
* @access private
* @private
*/
function _readsDone(status) {
/* jshint validthis: true */
@ -202,7 +230,7 @@ ClientReadableStream.prototype._readsDone = _readsDone;
/**
* Called to indicate that we have received a status from the server.
* @access private
* @private
*/
function _receiveStatus(status) {
/* jshint validthis: true */
@ -215,7 +243,7 @@ ClientReadableStream.prototype._receiveStatus = _receiveStatus;
/**
* If we have both processed all incoming messages and received the status from
* the server, emit the status. Otherwise, do nothing.
* @access private
* @private
*/
function _emitStatusIfDone() {
/* jshint validthis: true */
@ -242,7 +270,7 @@ ClientReadableStream.prototype._emitStatusIfDone = _emitStatusIfDone;
/**
* Read the next object from the stream.
* @access private
* @private
* @param {*} size Ignored because we use objectMode=true
*/
function _read(size) {
@ -300,16 +328,19 @@ util.inherits(ClientDuplexStream, Duplex);
/**
* A stream that the client can read from or write to. Used for calls with
* duplex streaming.
* @constructor
* @constructor grpc~ClientDuplexStream
* @extends external:Duplex
* @borrows module:src/client~ClientUnaryCall#cancel as
* module:src/client~ClientDuplexStream#cancel
* @borrows module:src/client~ClientUnaryCall#getPeer as
* module:src/client~ClientDuplexStream#getPeer
* @param {grpc.Call} call Call object to proxy
* @param {module:src/common~serialize=} [serialize=identity] Serialization
* @borrows grpc~ClientUnaryCall#cancel as grpc~ClientDuplexStream#cancel
* @borrows grpc~ClientUnaryCall#getPeer as grpc~ClientDuplexStream#getPeer
* @borrows grpc~ClientWritableStream#write as grpc~ClientDuplexStream#write
* @borrows grpc~ClientUnaryCall#event:metadata as
* grpc~ClientDuplexStream#metadata
* @borrows grpc~ClientUnaryCall#event:status as
* grpc~ClientDuplexStream#status
* @param {grpc.internal~Call} call Call object to proxy
* @param {grpc~serialize=} [serialize=identity] Serialization
* function for requests
* @param {module:src/common~deserialize=} [deserialize=identity]
* @param {grpc~deserialize=} [deserialize=identity]
* Deserialization function for responses
*/
function ClientDuplexStream(call, serialize, deserialize) {
@ -336,8 +367,9 @@ ClientDuplexStream.prototype._read = _read;
ClientDuplexStream.prototype._write = _write;
/**
* Cancel the ongoing call
* @alias module:src/client~ClientUnaryCall#cancel
* Cancel the ongoing call. Results in the call ending with a CANCELLED status,
* unless it has already ended with some other status.
* @alias grpc~ClientUnaryCall#cancel
*/
function cancel() {
/* jshint validthis: true */
@ -352,7 +384,7 @@ ClientDuplexStream.prototype.cancel = cancel;
/**
* Get the endpoint this call/stream is connected to.
* @return {string} The URI of the endpoint
* @alias module:src/client~ClientUnaryCall#getPeer
* @alias grpc~ClientUnaryCall#getPeer
*/
function getPeer() {
/* jshint validthis: true */
@ -368,33 +400,31 @@ ClientDuplexStream.prototype.getPeer = getPeer;
* Any client call type
* @typedef {(ClientUnaryCall|ClientReadableStream|
* ClientWritableStream|ClientDuplexStream)}
* module:src/client~Call
* grpc.Client~Call
*/
/**
* Options that can be set on a call.
* @typedef {Object} module:src/client~CallOptions
* @property {(date|number)} deadline The deadline for the entire call to
* complete. A value of Infinity indicates that no deadline should be set.
* @property {(string)} host Server hostname to set on the call. Only meaningful
* @typedef {Object} grpc.Client~CallOptions
* @property {grpc~Deadline} deadline The deadline for the entire call to
* complete.
* @property {string} host Server hostname to set on the call. Only meaningful
* if different from the server address used to construct the client.
* @property {module:src/client~Call} parent Parent call. Used in servers when
* @property {grpc.Client~Call} parent Parent call. Used in servers when
* making a call as part of the process of handling a call. Used to
* propagate some information automatically, as specified by
* propagate_flags.
* @property {number} propagate_flags Indicates which properties of a parent
* call should propagate to this call. Bitwise combination of flags in
* [grpc.propagate]{@link module:index.propagate}.
* @property {module:src/credentials~CallCredentials} credentials The
* credentials that should be used to make this particular call.
* {@link grpc.propagate}.
* @property {grpc.credentials~CallCredentials} credentials The credentials that
* should be used to make this particular call.
*/
/**
* 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.
* Get a call object built with the provided options.
* @access private
* @param {module:src/client~CallOptions=} options Options object.
* @param {grpc.Client~CallOptions=} options Options object.
*/
function getCall(channel, method, options) {
var deadline;
@ -422,14 +452,14 @@ function getCall(channel, method, options) {
/**
* A generic gRPC client. Primarily useful as a base class for generated clients
* @alias module:src/client.Client
* @memberof grpc
* @constructor
* @param {string} address Server address to connect to
* @param {module:src/credentials~ChannelCredentials} credentials Credentials to
* use to connect to the server
* @param {grpc~ChannelCredentials} credentials Credentials to use to connect to
* the server
* @param {Object} options Options to apply to channel creation
*/
var Client = exports.Client = function Client(address, credentials, options) {
function Client(address, credentials, options) {
if (!options) {
options = {};
}
@ -445,19 +475,13 @@ var Client = exports.Client = function Client(address, credentials, options) {
/* Private fields use $ as a prefix instead of _ because it is an invalid
* prefix of a method name */
this.$channel = new grpc.Channel(address, credentials, options);
};
}
exports.Client = Client;
/**
* @typedef {Error} module:src/client.Client~ServiceError
* @property {number} code The error code, a key of
* [grpc.status]{@link module:src/client.status}
* @property {module:metadata.Metadata} metadata Metadata sent with the status
* by the server, if any
*/
/**
* @callback module:src/client.Client~requestCallback
* @param {?module:src/client.Client~ServiceError} error The error, if the call
* @callback grpc.Client~requestCallback
* @param {?grpc~ServiceError} error The error, if the call
* failed
* @param {*} value The response value, if the call succeeded
*/
@ -466,17 +490,17 @@ var Client = exports.Client = function Client(address, credentials, options) {
* Make a unary request to the given method, using the given serialize
* and deserialize functions, with the given argument.
* @param {string} method The name of the method to request
* @param {module:src/common~serialize} serialize The serialization function for
* @param {grpc~serialize} serialize The serialization function for
* inputs
* @param {module:src/common~deserialize} deserialize The deserialization
* @param {grpc~deserialize} deserialize The deserialization
* function for outputs
* @param {*} argument The argument to the call. Should be serializable with
* serialize
* @param {module:src/metadata.Metadata=} metadata Metadata to add to the call
* @param {module:src/client~CallOptions=} options Options map
* @param {module:src/client.Client~requestCallback} callback The callback to
* @param {grpc.Metadata=} metadata Metadata to add to the call
* @param {grpc.Client~CallOptions=} options Options map
* @param {grpc.Client~requestCallback} callback The callback to
* for when the response is received
* @return {EventEmitter} An event emitter for stream related events
* @return {grpc~ClientUnaryCall} An event emitter for stream related events
*/
Client.prototype.makeUnaryRequest = function(method, serialize, deserialize,
argument, metadata, options,
@ -548,17 +572,17 @@ Client.prototype.makeUnaryRequest = function(method, serialize, deserialize,
* Make a client stream request to the given method, using the given serialize
* and deserialize functions, with the given argument.
* @param {string} method The name of the method to request
* @param {module:src/common~serialize} serialize The serialization function for
* @param {grpc~serialize} serialize The serialization function for
* inputs
* @param {module:src/common~deserialize} deserialize The deserialization
* @param {grpc~deserialize} deserialize The deserialization
* function for outputs
* @param {module:src/metadata.Metadata=} metadata Array of metadata key/value
* pairs to add to the call
* @param {module:src/client~CallOptions=} options Options map
* @param {Client~requestCallback} callback The callback to for when the
* @param {grpc.Metadata=} metadata Array of metadata key/value pairs to add to
* the call
* @param {grpc.Client~CallOptions=} options Options map
* @param {grpc.Client~requestCallback} callback The callback to for when the
* response is received
* @return {module:src/client~ClientWritableStream} An event emitter for stream
* related events
* @return {grpc~ClientWritableStream} An event emitter for stream related
* events
*/
Client.prototype.makeClientStreamRequest = function(method, serialize,
deserialize, metadata,
@ -631,17 +655,16 @@ Client.prototype.makeClientStreamRequest = function(method, serialize,
* Make a server stream request to the given method, with the given serialize
* and deserialize function, using the given argument
* @param {string} method The name of the method to request
* @param {module:src/common~serialize} serialize The serialization function for
* inputs
* @param {module:src/common~deserialize} deserialize The deserialization
* @param {grpc~serialize} serialize The serialization function for inputs
* @param {grpc~deserialize} deserialize The deserialization
* function for outputs
* @param {*} argument The argument to the call. Should be serializable with
* serialize
* @param {module:src/metadata.Metadata=} metadata Array of metadata key/value
* pairs to add to the call
* @param {module:src/client~CallOptions=} options Options map
* @return {module:src/client~ClientReadableStream} An event emitter for stream
* related events
* @param {grpc.Metadata=} metadata Array of metadata key/value pairs to add to
* the call
* @param {grpc.Client~CallOptions=} options Options map
* @return {grpc~ClientReadableStream} An event emitter for stream related
* events
*/
Client.prototype.makeServerStreamRequest = function(method, serialize,
deserialize, argument,
@ -693,15 +716,13 @@ Client.prototype.makeServerStreamRequest = function(method, serialize,
/**
* Make a bidirectional stream request with this method on the given channel.
* @param {string} method The name of the method to request
* @param {module:src/common~serialize} serialize The serialization function for
* inputs
* @param {module:src/common~deserialize} deserialize The deserialization
* @param {grpc~serialize} serialize The serialization function for inputs
* @param {grpc~deserialize} deserialize The deserialization
* function for outputs
* @param {module:src/metadata.Metadata=} metadata Array of metadata key/value
* @param {grpc.Metadata=} metadata Array of metadata key/value
* pairs to add to the call
* @param {module:src/client~CallOptions=} options Options map
* @return {module:src/client~ClientDuplexStream} An event emitter for stream
* related events
* @param {grpc.Client~CallOptions=} options Options map
* @return {grpc~ClientDuplexStream} An event emitter for stream related events
*/
Client.prototype.makeBidiStreamRequest = function(method, serialize,
deserialize, metadata,
@ -743,6 +764,9 @@ Client.prototype.makeBidiStreamRequest = function(method, serialize,
return stream;
};
/**
* Close this client.
*/
Client.prototype.close = function() {
this.$channel.close();
};
@ -761,8 +785,7 @@ Client.prototype.getChannel = function() {
* 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 {grpc~Deadline} deadline When to stop waiting for a connection.
* @param {function(Error)} callback The callback to call when done attempting
* to connect.
*/
@ -788,7 +811,7 @@ Client.prototype.waitForReady = function(deadline, callback) {
/**
* Map with short names for each of the requester maker functions. Used in
* makeClientConstructor
* @access private
* @private
*/
var requester_funcs = {
unary: Client.prototype.makeUnaryRequest,
@ -834,9 +857,15 @@ var deprecated_request_wrap = {
/**
* Creates a constructor for a client with the given methods, as specified in
* the methods argument.
* @param {module:src/common~ServiceDefinition} methods An object mapping
* method names to method attributes
* the methods argument. The resulting class will have an instance method for
* each method in the service, which is a partial application of one of the
* [Client]{@link grpc.Client} request methods, depending on `requestSerialize`
* and `responseSerialize`, with the `method`, `serialize`, and `deserialize`
* arguments predefined.
* @memberof grpc
* @alias grpc~makeGenericClientConstructor
* @param {grpc~ServiceDefinition} methods An object mapping method names to
* method attributes
* @param {string} serviceName The fully qualified name of the service
* @param {Object} class_options An options object.
* @param {boolean=} [class_options.deprecatedArgumentOrder=false] Indicates
@ -844,9 +873,8 @@ var deprecated_request_wrap = {
* arguments at the end instead of the callback at the end. This option
* is only a temporary stopgap measure to smooth an API breakage.
* It is deprecated, and new code should not use it.
* @return {function(string, Object)} New client constructor, which is a
* subclass of [grpc.Client]{@link module:src/client.Client}, and has the
* same arguments as that constructor.
* @return {function} New client constructor, which is a subclass of
* {@link grpc.Client}, and has the same arguments as that constructor.
*/
exports.makeClientConstructor = function(methods, serviceName,
class_options) {
@ -898,8 +926,11 @@ exports.makeClientConstructor = function(methods, serviceName,
/**
* Return the underlying channel object for the specified client
* @memberof grpc
* @alias grpc~getClientChannel
* @param {Client} client
* @return {Channel} The channel
* @see grpc.Client#getChannel
*/
exports.getClientChannel = function(client) {
return Client.prototype.getChannel.call(client);
@ -911,22 +942,15 @@ exports.getClientChannel = function(client) {
* 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.
* @memberof grpc
* @alias grpc~waitForClientReady
* @param {Client} client The client to wait on
* @param {(Date|Number)} deadline When to stop waiting for a connection. Pass
* @param {grpc~Deadline} 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.
* @see grpc.Client#waitForReady
*/
exports.waitForClientReady = function(client, deadline, callback) {
Client.prototype.waitForReady.call(client, deadline, callback);
};
/**
* Map of status code names to status codes
*/
exports.status = constants.status;
/**
* See docs for client.callError
*/
exports.callError = grpc.callError;

64
src/common.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2015, Google Inc.
* All rights reserved.
*
@ -31,12 +31,6 @@
*
*/
/**
* This module contains functions that are common to client and server
* code. None of them should be used directly by gRPC users.
* @module
*/
'use strict';
var _ = require('lodash');
@ -62,16 +56,19 @@ exports.wrapIgnoreNull = function wrapIgnoreNull(func) {
/**
* The logger object for the gRPC module. Defaults to console.
* @private
*/
exports.logger = console;
/**
* The current logging verbosity. 0 corresponds to logging everything
* @private
*/
exports.logVerbosity = 0;
/**
* Log a message if the severity is at least as high as the current verbosity
* @private
* @param {Number} severity A value of the grpc.logVerbosity map
* @param {String} message The message to log
*/
@ -83,6 +80,7 @@ exports.log = function log(severity, message) {
/**
* Default options for loading proto files into gRPC
* @alias grpc~defaultLoadOptions
*/
exports.defaultGrpcOptions = {
convertFieldsToCamelCase: false,
@ -94,6 +92,30 @@ exports.defaultGrpcOptions = {
// JSDoc definitions that are used in multiple other modules
/**
* Represents the status of a completed request. If `code` is
* {@link grpc.status}.OK, then the request has completed successfully.
* Otherwise, the request has failed, `details` will contain a description of
* the error. Either way, `metadata` contains the trailing response metadata
* sent by the server when it finishes processing the call.
* @typedef {object} grpc~StatusObject
* @property {number} code The error code, a key of {@link grpc.status}
* @property {string} details Human-readable description of the status
* @property {grpc.Metadata} metadata Trailing metadata sent with the status,
* if applicable
*/
/**
* Describes how a request has failed. The member `message` will be the same as
* `details` in {@link grpc~StatusObject}, and `code` and `metadata` are the
* same as in that object.
* @typedef {Error} grpc~ServiceError
* @property {number} code The error code, a key of {@link grpc.status} that is
* not `grpc.status.OK`
* @property {grpc.Metadata} metadata Trailing metadata sent with the status,
* if applicable
*/
/**
* The EventEmitter class in the event standard module
* @external EventEmitter
@ -120,38 +142,46 @@ exports.defaultGrpcOptions = {
/**
* A serialization function
* @callback module:src/common~serialize
* @callback grpc~serialize
* @param {*} value The value to serialize
* @return {Buffer} The value serialized as a byte sequence
*/
/**
* A deserialization function
* @callback module:src/common~deserialize
* @callback grpc~deserialize
* @param {Buffer} data The byte sequence to deserialize
* @return {*} The data deserialized as a value
*/
/**
* The deadline of an operation. If it is a date, the deadline is reached at
* the date and time specified. If it is a finite number, it is treated as
* a number of milliseconds since the Unix Epoch. If it is Infinity, the
* deadline will never be reached. If it is -Infinity, the deadline has already
* passed.
* @typedef {(number|date)} grpc~Deadline
*/
/**
* An object that completely defines a service method signature.
* @typedef {Object} module:src/common~MethodDefinition
* @typedef {Object} grpc~MethodDefinition
* @property {string} path The method's URL path
* @property {boolean} requestStream Indicates whether the method accepts
* a stream of requests
* @property {boolean} responseStream Indicates whether the method returns
* a stream of responses
* @property {module:src/common~serialize} requestSerialize Serialization
* @property {grpc~serialize} requestSerialize Serialization
* function for request values
* @property {module:src/common~serialize} responseSerialize Serialization
* @property {grpc~serialize} responseSerialize Serialization
* function for response values
* @property {module:src/common~deserialize} requestDeserialize Deserialization
* @property {grpc~deserialize} requestDeserialize Deserialization
* function for request data
* @property {module:src/common~deserialize} responseDeserialize Deserialization
* @property {grpc~deserialize} responseDeserialize Deserialization
* function for repsonse data
*/
/**
* An object that completely defines a service.
* @typedef {Object.<string, module:src/common~MethodDefinition>}
* module:src/common~ServiceDefinition
* @typedef {Object.<string, grpc~MethodDefinition>} grpc~ServiceDefinition
*/

24
src/constants.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2017, Google Inc.
* All rights reserved.
*
@ -31,16 +31,14 @@
*
*/
/**
* @module
*/
/* The comments about status codes are copied verbatim (with some formatting
* modifications) from include/grpc/impl/codegen/status.h, for the purpose of
* including them in generated documentation.
*/
/**
* Enum of status codes that gRPC can return
* @memberof grpc
* @alias grpc.status
* @readonly
* @enum {number}
*/
@ -178,6 +176,8 @@ exports.status = {
* Users are encouraged to write propagation masks as deltas from the default.
* i.e. write `grpc.propagate.DEFAULTS & ~grpc.propagate.DEADLINE` to disable
* deadline propagation.
* @memberof grpc
* @alias grpc.propagate
* @enum {number}
*/
exports.propagate = {
@ -194,9 +194,11 @@ exports.propagate = {
/**
* Call error constants. Call errors almost always indicate bugs in the gRPC
* library, and these error codes are mainly useful for finding those bugs.
* @memberof grpc
* @readonly
* @enum {number}
*/
exports.callError = {
const callError = {
OK: 0,
ERROR: 1,
NOT_ON_SERVER: 2,
@ -213,9 +215,14 @@ exports.callError = {
PAYLOAD_TYPE_MISMATCH: 14
};
exports.callError = callError;
/**
* Write flags: these can be bitwise or-ed to form write options that modify
* how data is written.
* @memberof grpc
* @alias grpc.writeFlags
* @readonly
* @enum {number}
*/
exports.writeFlags = {
@ -232,6 +239,9 @@ exports.writeFlags = {
};
/**
* @memberof grpc
* @alias grpc.logVerbosity
* @readonly
* @enum {number}
*/
exports.logVerbosity = {

63
src/credentials.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2015, Google Inc.
* All rights reserved.
*
@ -48,6 +48,7 @@
* For example, to create a client secured with SSL that uses Google
* default application credentials to authenticate:
*
* @example
* var channel_creds = credentials.createSsl(root_certs);
* (new GoogleAuth()).getApplicationDefault(function(err, credential) {
* var call_creds = credentials.createFromGoogleCredential(credential);
@ -56,15 +57,25 @@
* var client = new Client(address, combined_creds);
* });
*
* @module
* @namespace grpc.credentials
*/
'use strict';
var grpc = require('./grpc_extension');
/**
* This cannot be constructed directly. Instead, instances of this class should
* be created using the factory functions in {@link grpc.credentials}
* @constructor grpc.credentials~CallCredentials
*/
var CallCredentials = grpc.CallCredentials;
/**
* This cannot be constructed directly. Instead, instances of this class should
* be created using the factory functions in {@link grpc.credentials}
* @constructor grpc.credentials~ChannelCredentials
*/
var ChannelCredentials = grpc.ChannelCredentials;
var Metadata = require('./metadata.js');
@ -75,25 +86,49 @@ var constants = require('./constants');
var _ = require('lodash');
/**
* @external GoogleCredential
* @see https://github.com/google/google-auth-library-nodejs
*/
/**
* Create an SSL Credentials object. If using a client-side certificate, both
* the second and third arguments must be passed.
* @memberof grpc.credentials
* @alias grpc.credentials.createSsl
* @kind function
* @param {Buffer} root_certs The root certificate data
* @param {Buffer=} private_key The client certificate private key, if
* applicable
* @param {Buffer=} cert_chain The client certificate cert chain, if applicable
* @return {ChannelCredentials} The SSL Credentials object
* @return {grpc.credentials.ChannelCredentials} The SSL Credentials object
*/
exports.createSsl = ChannelCredentials.createSsl;
/**
* @callback grpc.credentials~metadataCallback
* @param {Error} error The error, if getting metadata failed
* @param {grpc.Metadata} metadata The metadata
*/
/**
* @callback grpc.credentials~generateMetadata
* @param {Object} params Parameters that can modify metadata generation
* @param {string} params.service_url The URL of the service that the call is
* going to
* @param {grpc.credentials~metadataCallback} callback
*/
/**
* Create a gRPC credentials object from a metadata generation function. This
* function gets the service URL and a callback as parameters. The error
* passed to the callback can optionally have a 'code' value attached to it,
* which corresponds to a status code that this library uses.
* @param {function(String, function(Error, Metadata))} metadata_generator The
* function that generates metadata
* @return {CallCredentials} The credentials object
* @memberof grpc.credentials
* @alias grpc.credentials.createFromMetadataGenerator
* @param {grpc.credentials~generateMetadata} metadata_generator The function
* that generates metadata
* @return {grpc.credentials.CallCredentials} The credentials object
*/
exports.createFromMetadataGenerator = function(metadata_generator) {
return CallCredentials.createFromPlugin(function(service_url, cb_data,
@ -119,8 +154,11 @@ exports.createFromMetadataGenerator = function(metadata_generator) {
/**
* Create a gRPC credential from a Google credential object.
* @param {Object} google_credential The Google credential object to use
* @return {CallCredentials} The resulting credentials object
* @memberof grpc.credentials
* @alias grpc.credentials.createFromGoogleCredential
* @param {external:GoogleCredential} google_credential The Google credential
* object to use
* @return {grpc.credentials.CallCredentials} The resulting credentials object
*/
exports.createFromGoogleCredential = function(google_credential) {
return exports.createFromMetadataGenerator(function(auth_context, callback) {
@ -141,6 +179,8 @@ exports.createFromGoogleCredential = function(google_credential) {
/**
* Combine a ChannelCredentials with any number of CallCredentials into a single
* ChannelCredentials object.
* @memberof grpc.credentials
* @alias grpc.credentials.combineChannelCredentials
* @param {ChannelCredentials} channel_credential The ChannelCredentials to
* start with
* @param {...CallCredentials} credentials The CallCredentials to compose
@ -157,6 +197,8 @@ exports.combineChannelCredentials = function(channel_credential) {
/**
* Combine any number of CallCredentials into a single CallCredentials object
* @memberof grpc.credentials
* @alias grpc.credentials.combineCallCredentials
* @param {...CallCredentials} credentials the CallCredentials to compose
* @return CallCredentials A credentials object that combines all of the input
* credentials
@ -172,6 +214,9 @@ exports.combineCallCredentials = function() {
/**
* Create an insecure credentials object. This is used to create a channel that
* does not use SSL. This cannot be composed with anything.
* @memberof grpc.credentials
* @alias grpc.credentials.createInsecure
* @kind function
* @return {ChannelCredentials} The insecure credentials object
*/
exports.createInsecure = ChannelCredentials.createInsecure;

4
src/grpc_extension.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2016, Google Inc.
* All rights reserved.
*

15
src/metadata.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2015, Google Inc.
* All rights reserved.
*
@ -31,15 +31,6 @@
*
*/
/**
* Metadata module
*
* This module defines the Metadata class, which represents header and trailer
* metadata for gRPC calls.
*
* @module
*/
'use strict';
var _ = require('lodash');
@ -48,8 +39,8 @@ var grpc = require('./grpc_extension');
/**
* Class for storing metadata. Keys are normalized to lowercase ASCII.
* @memberof grpc
* @constructor
* @alias module:src/metadata.Metadata
* @example
* var metadata = new metadata_module.Metadata();
* metadata.set('key1', 'value1');

9
src/protobuf_js_5_common.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2017, Google Inc.
* All rights reserved.
*
@ -31,6 +31,11 @@
*
*/
/**
* @module
* @private
*/
'use strict';
var _ = require('lodash');

9
src/protobuf_js_6_common.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2017, Google Inc.
* All rights reserved.
*
@ -31,6 +31,11 @@
*
*/
/**
* @module
* @private
*/
'use strict';
var _ = require('lodash');

453
src/server.js
View File

@ -1,5 +1,5 @@
/*
*
/**
* @license
* Copyright 2015, Google Inc.
* All rights reserved.
*
@ -31,22 +31,6 @@
*
*/
/**
* Server module
*
* This module contains all the server code for Node gRPC: both the Server
* class itself and the method handler code for all types of methods.
*
* For example, to create a Server, add a service, and start it:
*
* var server = new server_module.Server();
* server.addProtoService(protobuf_service_descriptor, service_implementation);
* server.bind('address:port', server_credential);
* server.start();
*
* @module
*/
'use strict';
var _ = require('lodash');
@ -70,9 +54,9 @@ 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
* @private
* @param {grpc.internal~Call} call The call to send the error on
* @param {(Object|Error)} error The error object
*/
function handleError(call, error) {
var statusMetadata = new Metadata();
@ -104,14 +88,14 @@ function handleError(call, error) {
/**
* Send a response to a unary or client streaming call.
* @access private
* @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
* @param {grpc~serialize} serialize Serialization function for the
* response
* @param {Metadata=} metadata Optional trailing metadata to send with status
* @param {number=} flags Flags for modifying how the message is sent.
* Defaults to 0.
* @param {grpc.Metadata=} metadata Optional trailing metadata to send with
* status
* @param {number=} [flags=0] Flags for modifying how the message is sent.
*/
function sendUnaryResponse(call, value, serialize, metadata, flags) {
var end_batch = {};
@ -146,7 +130,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
* @private
* @param {Writable} stream The stream to set up
* @param {function(*):Buffer=} Serialization function for responses
*/
@ -225,9 +209,9 @@ function setUpWritable(stream, serialize) {
/**
* Initialize a readable stream. This is used for both the readable and duplex
* stream constructors.
* @access private
* @private
* @param {Readable} stream The stream to initialize
* @param {function(Buffer):*=} deserialize Deserialization function for
* @param {grpc~deserialize} deserialize Deserialization function for
* incoming data.
*/
function setUpReadable(stream, deserialize) {
@ -245,34 +229,88 @@ function setUpReadable(stream, deserialize) {
});
}
/**
* Emitted when the call has been cancelled. After this has been emitted, the
* call's `cancelled` property will be set to `true`.
* @event grpc~ServerUnaryCall~cancelled
*/
util.inherits(ServerUnaryCall, EventEmitter);
function ServerUnaryCall(call) {
/**
* An EventEmitter. Used for unary calls.
* @constructor grpc~ServerUnaryCall
* @extends external:EventEmitter
* @param {grpc.internal~Call} call The call object associated with the request
* @param {grpc.Metadata} metadata The request metadata from the client
*/
function ServerUnaryCall(call, metadata) {
EventEmitter.call(this);
this.call = call;
/**
* Indicates if the call has been cancelled
* @member {boolean} grpc~ServerUnaryCall#cancelled
*/
this.cancelled = false;
/**
* The request metadata from the client
* @member {grpc.Metadata} grpc~ServerUnaryCall#metadata
*/
this.metadata = metadata;
/**
* The request message from the client
* @member {*} grpc~ServerUnaryCall#request
*/
this.request = undefined;
}
/**
* Emitted when the call has been cancelled. After this has been emitted, the
* call's `cancelled` property will be set to `true`.
* @event grpc~ServerWritableStream~cancelled
*/
util.inherits(ServerWritableStream, Writable);
/**
* A stream that the server can write to. Used for calls that are streaming from
* the server side.
* @constructor
* @param {grpc.Call} call The call object to send data with
* @param {function(*):Buffer=} serialize Serialization function for writes
* @constructor grpc~ServerWritableStream
* @extends external:Writable
* @borrows grpc~ServerUnaryCall#sendMetadata as
* grpc~ServerWritableStream#sendMetadata
* @borrows grpc~ServerUnaryCall#getPeer as grpc~ServerWritableStream#getPeer
* @param {grpc.internal~Call} call The call object to send data with
* @param {grpc.Metadata} metadata The request metadata from the client
* @param {grpc~serialize} serialize Serialization function for writes
*/
function ServerWritableStream(call, serialize) {
function ServerWritableStream(call, metadata, serialize) {
Writable.call(this, {objectMode: true});
this.call = call;
this.finished = false;
setUpWritable(this, serialize);
/**
* Indicates if the call has been cancelled
* @member {boolean} grpc~ServerWritableStream#cancelled
*/
this.cancelled = false;
/**
* The request metadata from the client
* @member {grpc.Metadata} grpc~ServerWritableStream#metadata
*/
this.metadata = metadata;
/**
* The request message from the client
* @member {*} grpc~ServerWritableStream#request
*/
this.request = undefined;
}
/**
* Start writing a chunk of data. This is an implementation of a method required
* for implementing stream.Writable.
* @access private
* @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
@ -312,19 +350,40 @@ function _write(chunk, encoding, callback) {
ServerWritableStream.prototype._write = _write;
/**
* Emitted when the call has been cancelled. After this has been emitted, the
* call's `cancelled` property will be set to `true`.
* @event grpc~ServerReadableStream~cancelled
*/
util.inherits(ServerReadableStream, Readable);
/**
* A stream that the server can read from. Used for calls that are streaming
* from the client side.
* @constructor
* @param {grpc.Call} call The call object to read data with
* @param {function(Buffer):*=} deserialize Deserialization function for reads
* @constructor grpc~ServerReadableStream
* @extends external:Readable
* @borrows grpc~ServerUnaryCall#sendMetadata as
* grpc~ServerReadableStream#sendMetadata
* @borrows grpc~ServerUnaryCall#getPeer as grpc~ServerReadableStream#getPeer
* @param {grpc.internal~Call} call The call object to read data with
* @param {grpc.Metadata} metadata The request metadata from the client
* @param {grpc~deserialize} deserialize Deserialization function for reads
*/
function ServerReadableStream(call, deserialize) {
function ServerReadableStream(call, metadata, deserialize) {
Readable.call(this, {objectMode: true});
this.call = call;
setUpReadable(this, deserialize);
/**
* Indicates if the call has been cancelled
* @member {boolean} grpc~ServerReadableStream#cancelled
*/
this.cancelled = false;
/**
* The request metadata from the client
* @member {grpc.Metadata} grpc~ServerReadableStream#metadata
*/
this.metadata = metadata;
}
/**
@ -381,22 +440,43 @@ function _read(size) {
ServerReadableStream.prototype._read = _read;
/**
* Emitted when the call has been cancelled. After this has been emitted, the
* call's `cancelled` property will be set to `true`.
* @event grpc~ServerDuplexStream~cancelled
*/
util.inherits(ServerDuplexStream, Duplex);
/**
* A stream that the server can read from or write to. Used for calls with
* duplex streaming.
* @constructor
* @param {grpc.Call} call Call object to proxy
* @param {function(*):Buffer=} serialize Serialization function for requests
* @param {function(Buffer):*=} deserialize Deserialization function for
* @constructor grpc~ServerDuplexStream
* @extends external:Duplex
* @borrows grpc~ServerUnaryCall#sendMetadata as
* grpc~ServerDuplexStream#sendMetadata
* @borrows grpc~ServerUnaryCall#getPeer as grpc~ServerDuplexStream#getPeer
* @param {grpc.internal~Call} call Call object to proxy
* @param {grpc.Metadata} metadata The request metadata from the client
* @param {grpc~serialize} serialize Serialization function for requests
* @param {grpc~deserialize} deserialize Deserialization function for
* responses
*/
function ServerDuplexStream(call, serialize, deserialize) {
function ServerDuplexStream(call, metadata, serialize, deserialize) {
Duplex.call(this, {objectMode: true});
this.call = call;
setUpWritable(this, serialize);
setUpReadable(this, deserialize);
/**
* Indicates if the call has been cancelled
* @member {boolean} grpc~ServerReadableStream#cancelled
*/
this.cancelled = false;
/**
* The request metadata from the client
* @member {grpc.Metadata} grpc~ServerReadableStream#metadata
*/
this.metadata = metadata;
}
ServerDuplexStream.prototype._read = _read;
@ -404,6 +484,7 @@ ServerDuplexStream.prototype._write = _write;
/**
* Send the initial metadata for a writable stream.
* @alias grpc~ServerUnaryCall#sendMetadata
* @param {Metadata} responseMetadata Metadata to send
*/
function sendMetadata(responseMetadata) {
@ -430,6 +511,7 @@ ServerDuplexStream.prototype.sendMetadata = sendMetadata;
/**
* Get the endpoint this call/stream is connected to.
* @alias grpc~ServerUnaryCall#getPeer
* @return {string} The URI of the endpoint
*/
function getPeer() {
@ -445,6 +527,7 @@ ServerDuplexStream.prototype.getPeer = getPeer;
/**
* Wait for the client to close, then emit a cancelled event if the client
* cancelled.
* @private
*/
function waitForCancel() {
/* jshint validthis: true */
@ -467,19 +550,42 @@ ServerReadableStream.prototype.waitForCancel = waitForCancel;
ServerWritableStream.prototype.waitForCancel = waitForCancel;
ServerDuplexStream.prototype.waitForCancel = waitForCancel;
/**
* Callback function passed to server handlers that handle methods with unary
* responses.
* @callback grpc.Server~sendUnaryData
* @param {grpc~ServiceError} error An error, if the call failed
* @param {*} value The response value. Must be a valid argument to the
* `responseSerialize` method of the method that is being handled
* @param {grpc.Metadata=} trailer Trailing metadata to send, if applicable
* @param {grpc.writeFlags=} flags Flags to modify writing the response
*/
/**
* User-provided method to handle unary requests on a server
* @callback grpc.Server~handleUnaryCall
* @param {grpc~ServerUnaryCall} call The call object
* @param {grpc.Server~sendUnaryData} callback The callback to call to respond
* to the request
*/
/**
* Fully handle a unary call
* @access private
* @param {grpc.Call} call The call to handle
* @private
* @param {grpc.internal~Call} call The call to handle
* @param {Object} handler Request handler object for the method that was called
* @param {Metadata} metadata Metadata from the client
* @param {grpc~Server.handleUnaryCall} handler.func The handler function
* @param {grpc~deserialize} handler.deserialize The deserialization function
* for request data
* @param {grpc~serialize} handler.serialize The serialization function for
* response data
* @param {grpc.Metadata} metadata Metadata from the client
*/
function handleUnary(call, handler, metadata) {
var emitter = new ServerUnaryCall(call);
var emitter = new ServerUnaryCall(call, metadata);
emitter.on('error', function(error) {
handleError(call, error);
});
emitter.metadata = metadata;
emitter.waitForCancel();
var batch = {};
batch[grpc.opType.RECV_MESSAGE] = true;
@ -511,17 +617,28 @@ function handleUnary(call, handler, metadata) {
});
}
/**
* User provided method to handle server streaming methods on the server.
* @callback grpc.Server~handleServerStreamingCall
* @param {grpc~ServerWritableStream} call The call object
*/
/**
* Fully handle a server streaming call
* @access private
* @param {grpc.Call} call The call to handle
* @private
* @param {grpc.internal~Call} call The call to handle
* @param {Object} handler Request handler object for the method that was called
* @param {Metadata} metadata Metadata from the client
* @param {grpc~Server.handleServerStreamingCall} handler.func The handler
* function
* @param {grpc~deserialize} handler.deserialize The deserialization function
* for request data
* @param {grpc~serialize} handler.serialize The serialization function for
* response data
* @param {grpc.Metadata} metadata Metadata from the client
*/
function handleServerStreaming(call, handler, metadata) {
var stream = new ServerWritableStream(call, handler.serialize);
var stream = new ServerWritableStream(call, metadata, handler.serialize);
stream.waitForCancel();
stream.metadata = metadata;
var batch = {};
batch[grpc.opType.RECV_MESSAGE] = true;
call.startBatch(batch, function(err, result) {
@ -540,20 +657,33 @@ function handleServerStreaming(call, handler, metadata) {
});
}
/**
* User provided method to handle client streaming methods on the server.
* @callback grpc.Server~handleClientStreamingCall
* @param {grpc~ServerReadableStream} call The call object
* @param {grpc.Server~sendUnaryData} callback The callback to call to respond
* to the request
*/
/**
* Fully handle a client streaming call
* @access private
* @param {grpc.Call} call The call to handle
* @param {grpc.internal~Call} call The call to handle
* @param {Object} handler Request handler object for the method that was called
* @param {Metadata} metadata Metadata from the client
* @param {grpc~Server.handleClientStreamingCall} handler.func The handler
* function
* @param {grpc~deserialize} handler.deserialize The deserialization function
* for request data
* @param {grpc~serialize} handler.serialize The serialization function for
* response data
* @param {grpc.Metadata} metadata Metadata from the client
*/
function handleClientStreaming(call, handler, metadata) {
var stream = new ServerReadableStream(call, handler.deserialize);
var stream = new ServerReadableStream(call, metadata, handler.deserialize);
stream.on('error', function(error) {
handleError(call, error);
});
stream.waitForCancel();
stream.metadata = metadata;
handler.func(stream, function(err, value, trailer, flags) {
stream.terminate();
if (err) {
@ -567,18 +697,29 @@ function handleClientStreaming(call, handler, metadata) {
});
}
/**
* User provided method to handle bidirectional streaming calls on the server.
* @callback grpc.Server~handleBidiStreamingCall
* @param {grpc~ServerDuplexStream} call The call object
*/
/**
* Fully handle a bidirectional streaming call
* @access private
* @param {grpc.Call} call The call to handle
* @private
* @param {grpc.internal~Call} call The call to handle
* @param {Object} handler Request handler object for the method that was called
* @param {grpc~Server.handleBidiStreamingCall} handler.func The handler
* function
* @param {grpc~deserialize} handler.deserialize The deserialization function
* for request data
* @param {grpc~serialize} handler.serialize The serialization function for
* response data
* @param {Metadata} metadata Metadata from the client
*/
function handleBidiStreaming(call, handler, metadata) {
var stream = new ServerDuplexStream(call, handler.serialize,
var stream = new ServerDuplexStream(call, metadata, handler.serialize,
handler.deserialize);
stream.waitForCancel();
stream.metadata = metadata;
handler.func(stream);
}
@ -592,96 +733,90 @@ var streamHandlers = {
/**
* Constructs a server object that stores request handlers and delegates
* incoming requests to those handlers
* @memberof grpc
* @constructor
* @param {Object=} options Options that should be passed to the internal server
* implementation
* @example
* var server = new grpc.Server();
* server.addProtoService(protobuf_service_descriptor, service_implementation);
* server.bind('address:port', server_credential);
* server.start();
*/
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.start = function() {
if (this.started) {
throw new Error('Server is already running');
}
this.started = true;
server.start();
/**
* Handles the SERVER_RPC_NEW event. If there is a handler associated with
* the requested method, use that handler to respond to the request. Then
* wait for the next request
* @param {grpc.Event} event The event to handle with tag SERVER_RPC_NEW
*/
function handleNewCall(err, event) {
if (err) {
return;
}
var details = event.new_call;
var call = details.call;
var method = details.method;
var metadata = Metadata._fromCoreRepresentation(details.metadata);
if (method === null) {
return;
}
server.requestCall(handleNewCall);
var handler;
if (handlers.hasOwnProperty(method)) {
handler = handlers[method];
} else {
var batch = {};
batch[grpc.opType.SEND_INITIAL_METADATA] =
(new Metadata())._getCoreRepresentation();
batch[grpc.opType.SEND_STATUS_FROM_SERVER] = {
code: constants.status.UNIMPLEMENTED,
details: '',
metadata: {}
};
batch[grpc.opType.RECV_CLOSE_ON_SERVER] = true;
call.startBatch(batch, function() {});
return;
}
streamHandlers[handler.type](call, handler, metadata);
}
server.requestCall(handleNewCall);
};
/**
* Gracefully shuts down the server. The server will stop receiving new calls,
* and any pending calls will complete. The callback will be called when all
* pending calls have completed and the server is fully shut down. This method
* is idempotent with itself and forceShutdown.
* @param {function()} callback The shutdown complete callback
*/
this.tryShutdown = function(callback) {
server.tryShutdown(callback);
};
/**
* Forcibly shuts down the server. The server will stop receiving new calls
* and cancel all pending calls. When it returns, the server has shut down.
* This method is idempotent with itself and tryShutdown, and it will trigger
* any outstanding tryShutdown callbacks.
*/
this.forceShutdown = function() {
server.forceShutdown();
};
}
/**
* Start the server and begin handling requests
*/
Server.prototype.start = function() {
if (this.started) {
throw new Error('Server is already running');
}
var self = this;
this.started = true;
this._server.start();
/**
* Handles the SERVER_RPC_NEW event. If there is a handler associated with
* the requested method, use that handler to respond to the request. Then
* wait for the next request
* @param {grpc.internal~Event} event The event to handle with tag
* SERVER_RPC_NEW
*/
function handleNewCall(err, event) {
if (err) {
return;
}
var details = event.new_call;
var call = details.call;
var method = details.method;
var metadata = Metadata._fromCoreRepresentation(details.metadata);
if (method === null) {
return;
}
self._server.requestCall(handleNewCall);
var handler;
if (self.handlers.hasOwnProperty(method)) {
handler = self.handlers[method];
} else {
var batch = {};
batch[grpc.opType.SEND_INITIAL_METADATA] =
(new Metadata())._getCoreRepresentation();
batch[grpc.opType.SEND_STATUS_FROM_SERVER] = {
code: constants.status.UNIMPLEMENTED,
details: '',
metadata: {}
};
batch[grpc.opType.RECV_CLOSE_ON_SERVER] = true;
call.startBatch(batch, function() {});
return;
}
streamHandlers[handler.type](call, handler, metadata);
}
this._server.requestCall(handleNewCall);
};
/**
* Unified type for application handlers for all types of calls
* @typedef {(grpc.Server~handleUnaryCall
* |grpc.Server~handleClientStreamingCall
* |grpc.Server~handleServerStreamingCall
* |grpc.Server~handleBidiStreamingCall)} grpc.Server~handleCall
*/
/**
* Registers a handler to handle the named method. Fails if there already is
* a handler for the given method. Returns true on success
* @param {string} name The name of the method that the provided function should
* handle/respond to.
* @param {function} handler Function that takes a stream of request values and
* returns a stream of response values
* @param {function(*):Buffer} serialize Serialization function for responses
* @param {function(Buffer):*} deserialize Deserialization function for requests
* @param {grpc.Server~handleCall} handler Function that takes a stream of
* request values and returns a stream of response values
* @param {grpc~serialize} serialize Serialization function for responses
* @param {grpc~deserialize} deserialize Deserialization function for requests
* @param {string} type The streaming type of method that this handles
* @return {boolean} True if the handler was set. False if a handler was already
* set for that name.
@ -700,6 +835,27 @@ Server.prototype.register = function(name, handler, serialize, deserialize,
return true;
};
/**
* Gracefully shuts down the server. The server will stop receiving new calls,
* and any pending calls will complete. The callback will be called when all
* pending calls have completed and the server is fully shut down. This method
* is idempotent with itself and forceShutdown.
* @param {function()} callback The shutdown complete callback
*/
Server.prototype.tryShutdown = function(callback) {
this._server.tryShutdown(callback);
};
/**
* Forcibly shuts down the server. The server will stop receiving new calls
* and cancel all pending calls. When it returns, the server has shut down.
* This method is idempotent with itself and tryShutdown, and it will trigger
* any outstanding tryShutdown callbacks.
*/
Server.prototype.forceShutdown = function() {
this._server.forceShutdown();
};
var unimplementedStatusResponse = {
code: constants.status.UNIMPLEMENTED,
details: 'The server does not implement this method'
@ -721,13 +877,10 @@ var defaultHandler = {
};
/**
* 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.
* Add a service to the server, with a corresponding implementation.
* @param {grpc~ServiceDefinition} service The service descriptor
* @param {Object<String, grpc.Server~handleCall>} implementation Map of method
* names to method implementation for the provided service.
*/
Server.prototype.addService = function(service, implementation) {
if (!_.isObject(service) || !_.isObject(implementation)) {
@ -783,10 +936,10 @@ Server.prototype.addService = function(service, implementation) {
/**
* Add a proto service to the server, with a corresponding implementation
* @deprecated Use grpc.load and Server#addService instead
* @deprecated Use {@link grpc.Server#addService} instead
* @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.
* @param {Object<String, grpc.Server~handleCall>} implementation Map of method
* names to method implementation for the provided service.
*/
Server.prototype.addProtoService = function(service, implementation) {
var options;
@ -811,10 +964,11 @@ Server.prototype.addProtoService = function(service, implementation) {
};
/**
* Binds the server to the given port, with SSL enabled if creds is given
* Binds the server to the given port, with SSL disabled if creds is an
* insecure credentials object
* @param {string} port The port that the server should bind on, in the format
* "address:port"
* @param {ServerCredentials=} creds Server credential object to be used for
* @param {grpc.ServerCredentials} creds Server credential object to be used for
* SSL. Pass an insecure credentials object for an insecure port.
*/
Server.prototype.bind = function(port, creds) {
@ -824,7 +978,4 @@ Server.prototype.bind = function(port, creds) {
return this._server.addHttp2Port(port, creds);
};
/**
* @see module:src/server~Server
*/
exports.Server = Server;

8
test/surface_test.js
View File

@ -1330,14 +1330,14 @@ describe('Cancelling surface client', function() {
});
it('Should correctly cancel a unary call', function(done) {
var call = client.div({'divisor': 0, 'dividend': 0}, function(err, resp) {
assert.strictEqual(err.code, surface_client.status.CANCELLED);
assert.strictEqual(err.code, grpc.status.CANCELLED);
done();
});
call.cancel();
});
it('Should correctly cancel a client stream call', function(done) {
var call = client.sum(function(err, resp) {
assert.strictEqual(err.code, surface_client.status.CANCELLED);
assert.strictEqual(err.code, grpc.status.CANCELLED);
done();
});
call.cancel();
@ -1346,7 +1346,7 @@ describe('Cancelling surface client', function() {
var call = client.fib({'limit': 5});
call.on('data', function() {});
call.on('error', function(error) {
assert.strictEqual(error.code, surface_client.status.CANCELLED);
assert.strictEqual(error.code, grpc.status.CANCELLED);
done();
});
call.cancel();
@ -1355,7 +1355,7 @@ describe('Cancelling surface client', function() {
var call = client.divMany();
call.on('data', function() {});
call.on('error', function(error) {
assert.strictEqual(error.code, surface_client.status.CANCELLED);
assert.strictEqual(error.code, grpc.status.CANCELLED);
done();
});
call.cancel();