/* * * Copyright 2014, Google Inc. * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are * met: * * * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * * Redistributions in binary form must reproduce the above * copyright notice, this list of conditions and the following disclaimer * in the documentation and/or other materials provided with the * distribution. * * Neither the name of Google Inc. nor the names of its * contributors may be used to endorse or promote products derived from * this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * */ var _ = require('underscore'); var client = require('./client.js'); var common = require('./common.js'); var EventEmitter = require('events').EventEmitter; var stream = require('stream'); var Readable = stream.Readable; var Writable = stream.Writable; var Duplex = stream.Duplex; var util = require('util'); function forwardEvent(fromEmitter, toEmitter, event) { fromEmitter.on(event, function forward() { _.partial(toEmitter.emit, event).apply(toEmitter, arguments); }); } util.inherits(ClientReadableObjectStream, Readable); /** * Class for representing a gRPC server streaming call as a Node stream on the * client side. Extends from stream.Readable. * @constructor * @param {stream} stream Underlying binary Duplex stream for the call * @param {function(Buffer)} deserialize Function for deserializing binary data * @param {object} options Stream options */ function ClientReadableObjectStream(stream, deserialize, options) { options = _.extend(options, {objectMode: true}); Readable.call(this, options); this._stream = stream; var self = this; forwardEvent(stream, this, 'status'); forwardEvent(stream, this, 'metadata'); this._stream.on('data', function forwardData(chunk) { if (!self.push(deserialize(chunk))) { self._stream.pause(); } }); this._stream.pause(); } util.inherits(ClientWritableObjectStream, Writable); /** * Class for representing a gRPC client streaming call as a Node stream on the * client side. Extends from stream.Writable. * @constructor * @param {stream} stream Underlying binary Duplex stream for the call * @param {function(*):Buffer} serialize Function for serializing objects * @param {object} options Stream options */ function ClientWritableObjectStream(stream, serialize, options) { options = _.extend(options, {objectMode: true}); Writable.call(this, options); this._stream = stream; this._serialize = serialize; forwardEvent(stream, this, 'status'); forwardEvent(stream, this, 'metadata'); this.on('finish', function() { this._stream.end(); }); } util.inherits(ClientBidiObjectStream, Duplex); /** * Class for representing a gRPC bidi streaming call as a Node stream on the * client side. Extends from stream.Duplex. * @constructor * @param {stream} stream Underlying binary Duplex stream for the call * @param {function(*):Buffer} serialize Function for serializing objects * @param {function(Buffer)} deserialize Function for deserializing binary data * @param {object} options Stream options */ function ClientBidiObjectStream(stream, serialize, deserialize, options) { options = _.extend(options, {objectMode: true}); Duplex.call(this, options); this._stream = stream; this._serialize = serialize; var self = this; forwardEvent(stream, this, 'status'); forwardEvent(stream, this, 'metadata'); this._stream.on('data', function forwardData(chunk) { if (!self.push(deserialize(chunk))) { self._stream.pause(); } }); this._stream.pause(); this.on('finish', function() { this._stream.end(); }); } /** * _read implementation for both types of streams that allow reading. * @this {ClientReadableObjectStream|ClientBidiObjectStream} * @param {number} size Ignored */ function _read(size) { this._stream.resume(); } /** * See docs for _read */ ClientReadableObjectStream.prototype._read = _read; /** * See docs for _read */ ClientBidiObjectStream.prototype._read = _read; /** * _write implementation for both types of streams that allow writing * @this {ClientWritableObjectStream|ClientBidiObjectStream} * @param {*} chunk The value to write to the stream * @param {string} encoding Ignored * @param {function(Error)} callback Callback to call when finished writing */ function _write(chunk, encoding, callback) { this._stream.write(this._serialize(chunk), encoding, callback); } /** * See docs for _write */ ClientWritableObjectStream.prototype._write = _write; /** * See docs for _write */ ClientBidiObjectStream.prototype._write = _write; /** * 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 * @param {function(Buffer)} deserialize The deserialization function for * outputs * @return {Function} makeUnaryRequest */ function makeUnaryRequestFunction(method, serialize, deserialize) { /** * Make a unary request with this method on the given channel with the given * argument, callback, etc. * @this {SurfaceClient} Client object. Must have a channel member. * @param {*} argument The argument to the call. Should be serializable with * serialize * @param {function(?Error, value=)} callback The callback to for when the * response is received * @param {array=} metadata Array of metadata key/value pairs to add to the * call * @param {(number|Date)=} deadline The deadline for processing this request. * Defaults to infinite future * @return {EventEmitter} An event emitter for stream related events */ function makeUnaryRequest(argument, callback, metadata, deadline) { var stream = client.makeRequest(this.channel, method, metadata, deadline); var emitter = new EventEmitter(); forwardEvent(stream, emitter, 'status'); forwardEvent(stream, emitter, 'metadata'); stream.write(serialize(argument)); stream.end(); stream.on('data', function forwardData(chunk) { try { callback(null, deserialize(chunk)); } catch (e) { callback(e); } }); return emitter; } return makeUnaryRequest; } /** * Get a function that can make client stream requests to the specified method. * @param {string} method The name of the method to request * @param {function(*):Buffer} serialize The serialization function for inputs * @param {function(Buffer)} deserialize The deserialization function for * outputs * @return {Function} makeClientStreamRequest */ function makeClientStreamRequestFunction(method, serialize, deserialize) { /** * Make a client stream request with this method on the given channel with the * given callback, etc. * @this {SurfaceClient} Client object. Must have a channel member. * @param {function(?Error, value=)} callback The callback to for when the * response is received * @param {array=} metadata Array of metadata key/value pairs to add to the * call * @param {(number|Date)=} deadline The deadline for processing this request. * Defaults to infinite future * @return {EventEmitter} An event emitter for stream related events */ function makeClientStreamRequest(callback, metadata, deadline) { var stream = client.makeRequest(this.channel, method, metadata, deadline); var obj_stream = new ClientWritableObjectStream(stream, serialize, {}); stream.on('data', function forwardData(chunk) { try { callback(null, deserialize(chunk)); } catch (e) { callback(e); } }); return obj_stream; } return makeClientStreamRequest; } /** * Get a function that can make server stream requests to the specified method. * @param {string} method The name of the method to request * @param {function(*):Buffer} serialize The serialization function for inputs * @param {function(Buffer)} deserialize The deserialization function for * outputs * @return {Function} makeServerStreamRequest */ function makeServerStreamRequestFunction(method, serialize, deserialize) { /** * Make a server stream request with this method on the given channel with the * given argument, etc. * @this {SurfaceClient} Client object. Must have a channel member. * @param {*} argument The argument to the call. Should be serializable with * serialize * @param {array=} metadata Array of metadata key/value pairs to add to the * call * @param {(number|Date)=} deadline The deadline for processing this request. * Defaults to infinite future * @return {EventEmitter} An event emitter for stream related events */ function makeServerStreamRequest(argument, metadata, deadline) { var stream = client.makeRequest(this.channel, method, metadata, deadline); var obj_stream = new ClientReadableObjectStream(stream, deserialize, {}); stream.write(serialize(argument)); stream.end(); return obj_stream; } return makeServerStreamRequest; } /** * Get a function that can make bidirectional stream requests to the specified * method. * @param {string} method The name of the method to request * @param {function(*):Buffer} serialize The serialization function for inputs * @param {function(Buffer)} deserialize The deserialization function for * outputs * @return {Function} makeBidiStreamRequest */ function makeBidiStreamRequestFunction(method, serialize, deserialize) { /** * Make a bidirectional stream request with this method on the given channel. * @this {SurfaceClient} Client object. Must have a channel member. * @param {array=} metadata Array of metadata key/value pairs to add to the * call * @param {(number|Date)=} deadline The deadline for processing this request. * Defaults to infinite future * @return {EventEmitter} An event emitter for stream related events */ function makeBidiStreamRequest(metadata, deadline) { var stream = client.makeRequest(this.channel, method, metadata, deadline); var obj_stream = new ClientBidiObjectStream(stream, serialize, deserialize, {}); return obj_stream; } return makeBidiStreamRequest; } /** * Map with short names for each of the requester maker functions. Used in * makeClientConstructor */ var requester_makers = { unary: makeUnaryRequestFunction, server_stream: makeServerStreamRequestFunction, client_stream: makeClientStreamRequestFunction, bidi: makeBidiStreamRequestFunction } /** * Creates a constructor for clients for the given service * @param {ProtoBuf.Reflect.Service} service The service to generate a client * for * @return {function(string, Object)} New client constructor */ function makeClientConstructor(service) { var prefix = '/' + common.fullyQualifiedName(service) + '/'; /** * Create a client with the given methods * @constructor * @param {string} address The address of the server to connect to * @param {Object} options Options to pass to the underlying channel */ function SurfaceClient(address, options) { this.channel = new client.Channel(address, options); } _.each(service.children, function(method) { var method_type; if (method.requestStream) { if (method.responseStream) { method_type = 'bidi'; } else { method_type = 'client_stream'; } } else { if (method.responseStream) { method_type = 'server_stream'; } else { method_type = 'unary'; } } SurfaceClient.prototype[method.name] = requester_makers[method_type]( prefix + method.name, common.serializeCls(method.resolvedRequestType.build()), common.deserializeCls(method.resolvedResponseType.build())); }); SurfaceClient.service = service; return SurfaceClient; } exports.makeClientConstructor = makeClientConstructor; /** * See docs for client.status */ exports.status = client.status; /** * See docs for client.callError */ exports.callError = client.callError;