| /* |
| * |
| * 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 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. |
| * @param {client.Channel} channel The channel on which to make the request |
| * @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(channel, argument, callback, metadata, deadline) { |
| var stream = client.makeRequest(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. |
| * @param {client.Channel} channel The channel on which to make the request |
| * @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(channel, callback, metadata, deadline) { |
| var stream = client.makeRequest(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. |
| * @param {client.Channel} channel The channel on which to make the request |
| * @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(channel, argument, metadata, deadline) { |
| var stream = client.makeRequest(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. |
| * @param {client.Channel} channel The channel on which to make the request |
| * @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(channel, metadata, deadline) { |
| var stream = client.makeRequest(channel, method, metadata, deadline); |
| var obj_stream = new ClientBidiObjectStream(stream, |
| serialize, |
| deserialize, |
| {}); |
| return obj_stream; |
| } |
| return makeBidiStreamRequest; |
| } |
| |
| /** |
| * See docs for makeUnaryRequestFunction |
| */ |
| exports.makeUnaryRequestFunction = makeUnaryRequestFunction; |
| |
| /** |
| * See docs for makeClientStreamRequestFunction |
| */ |
| exports.makeClientStreamRequestFunction = makeClientStreamRequestFunction; |
| |
| /** |
| * See docs for makeServerStreamRequestFunction |
| */ |
| exports.makeServerStreamRequestFunction = makeServerStreamRequestFunction; |
| |
| /** |
| * See docs for makeBidiStreamRequestFunction |
| */ |
| exports.makeBidiStreamRequestFunction = makeBidiStreamRequestFunction; |
| |
| /** |
| * See docs for client.Channel |
| */ |
| exports.Channel = client.Channel; |
| /** |
| * See docs for client.status |
| */ |
| exports.status = client.status; |
| /** |
| * See docs for client.callError |
| */ |
| exports.callError = client.callError; |