legacy-libs/grpc/src/client.js

   1 /**
   2  * @license
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 /**
  20  * Client module
  21  *
  22  * This module contains the factory method for creating Client classes, and the
  23  * method calling code for all types of methods.
  24  *
  25  * @example <caption>Create a client and call a method on it</caption>
  26  *
  27  * var proto_obj = grpc.load(proto_file_path);
  28  * var Client = proto_obj.package.subpackage.ServiceName;
  29  * var client = new Client(server_address, client_credentials);
  30  * var call = client.unaryMethod(arguments, callback);
  31  */
  32
  33 'use strict';
  34
  35 var client_interceptors = require('./client_interceptors');
  36 var grpc = require('./grpc_extension');
  37
  38 var common = require('./common');
  39
  40 var Metadata = require('./metadata');
  41
  42 var constants = require('./constants');
  43
  44 var EventEmitter = require('events').EventEmitter;
  45
  46 var stream = require('stream');
  47
  48 var Readable = stream.Readable;
  49 var Writable = stream.Writable;
  50 var Duplex = stream.Duplex;
  51 var methodTypes = constants.methodTypes;
  52 var util = require('util');
  53 var version = require('../package.json').version;
  54
  55 /**
  56  * Initial response metadata sent by the server when it starts processing the
  57  * call
  58  * @event grpc~ClientUnaryCall#metadata
  59  * @type {grpc.Metadata}
  60  */
  61
  62 /**
  63  * Status of the call when it has completed.
  64  * @event grpc~ClientUnaryCall#status
  65  * @type grpc~StatusObject
  66  */
  67
  68 util.inherits(ClientUnaryCall, EventEmitter);
  69
  70 /**
  71  * An EventEmitter. Used for unary calls.
  72  * @constructor grpc~ClientUnaryCall
  73  * @extends external:EventEmitter
  74  * @param {grpc.internal~Call} call The call object associated with the request
  75  */
  76 function ClientUnaryCall(call) {
  77   EventEmitter.call(this);
  78   this.call = call;
  79 }
  80
  81 util.inherits(ClientWritableStream, Writable);
  82
  83 /**
  84  * A stream that the client can write to. Used for calls that are streaming from
  85  * the client side.
  86  * @constructor grpc~ClientWritableStream
  87  * @extends external:Writable
  88  * @borrows grpc~ClientUnaryCall#cancel as grpc~ClientWritableStream#cancel
  89  * @borrows grpc~ClientUnaryCall#getPeer as grpc~ClientWritableStream#getPeer
  90  * @borrows grpc~ClientUnaryCall#event:metadata as
  91  *     grpc~ClientWritableStream#metadata
  92  * @borrows grpc~ClientUnaryCall#event:status as
  93  *     grpc~ClientWritableStream#status
  94  * @param {InterceptingCall} call Exposes gRPC request operations, processed by
  95  *     an interceptor stack.
  96  */
  97 function ClientWritableStream(call) {
  98   Writable.call(this, {objectMode: true});
  99   this.call = call;
 100   var self = this;
 101   this.on('finish', function() {
 102     self.call.halfClose();
 103   });
 104 }
 105
 106 /**
 107  * Write a message to the request stream. If serializing the argument fails,
 108  * the call will be cancelled and the stream will end with an error.
 109  * @name grpc~ClientWritableStream#write
 110  * @kind function
 111  * @override
 112  * @param {*} message The message to write. Must be a valid argument to the
 113  *     serialize function of the corresponding method
 114  * @param {grpc.writeFlags} flags Flags to modify how the message is written
 115  * @param {Function} callback Callback for when this chunk of data is flushed
 116  * @return {boolean} As defined for [Writable]{@link external:Writable}
 117  */
 118
 119 /**
 120  * Attempt to write the given chunk. Calls the callback when done. This is an
 121  * implementation of a method needed for implementing stream.Writable.
 122  * @private
 123  * @param {*} chunk The chunk to write
 124  * @param {grpc.writeFlags} encoding Used to pass write flags
 125  * @param {function(Error=)} callback Called when the write is complete
 126  */
 127 function _write(chunk, encoding, callback) {
 128   /* jshint validthis: true */
 129   var self = this;
 130   if (this.writeFailed) {
 131     /* Once a write fails, just call the callback immediately to let the caller
 132        flush any pending writes. */
 133     setImmediate(callback);
 134     return;
 135   }
 136   var outerCallback = function(err, event) {
 137     if (err) {
 138       /* Assume that the call is complete and that writing failed because a
 139          status was received. In that case, set a flag to discard all future
 140          writes */
 141       self.writeFailed = true;
 142     }
 143     callback();
 144   };
 145   var context = {
 146     encoding: encoding,
 147     callback: outerCallback
 148   };
 149   this.call.sendMessageWithContext(context, chunk);
 150 }
 151
 152 ClientWritableStream.prototype._write = _write;
 153
 154 util.inherits(ClientReadableStream, Readable);
 155
 156 /**
 157  * A stream that the client can read from. Used for calls that are streaming
 158  * from the server side.
 159  * @constructor grpc~ClientReadableStream
 160  * @extends external:Readable
 161  * @borrows grpc~ClientUnaryCall#cancel as grpc~ClientReadableStream#cancel
 162  * @borrows grpc~ClientUnaryCall#getPeer as grpc~ClientReadableStream#getPeer
 163  * @borrows grpc~ClientUnaryCall#event:metadata as
 164  *     grpc~ClientReadableStream#metadata
 165  * @borrows grpc~ClientUnaryCall#event:status as
 166  *     grpc~ClientReadableStream#status
 167  * @param {InterceptingCall} call Exposes gRPC request operations, processed by
 168  *     an interceptor stack.
 169  */
 170 function ClientReadableStream(call) {
 171   Readable.call(this, {objectMode: true});
 172   this.call = call;
 173   this.finished = false;
 174   this.reading = false;
 175   /* Status generated from reading messages from the server. Overrides the
 176    * status from the server if not OK */
 177   this.read_status = null;
 178   /* Status received from the server. */
 179   this.received_status = null;
 180 }
 181
 182 /**
 183  * Called when all messages from the server have been processed. The status
 184  * parameter indicates that the call should end with that status. status
 185  * defaults to OK if not provided.
 186  * @param {Object!} status The status that the call should end with
 187  * @private
 188  */
 189 function _readsDone(status) {
 190   /* jshint validthis: true */
 191   if (!status) {
 192     status = {code: constants.status.OK, details: 'OK'};
 193   }
 194   if (status.code !== constants.status.OK) {
 195     this.call.cancelWithStatus(status.code, status.details);
 196   }
 197   this.finished = true;
 198   this.read_status = status;
 199   this._emitStatusIfDone();
 200 }
 201
 202 ClientReadableStream.prototype._readsDone = _readsDone;
 203
 204 /**
 205  * Called to indicate that we have received a status from the server.
 206  * @private
 207  */
 208 function _receiveStatus(status) {
 209   /* jshint validthis: true */
 210   this.received_status = status;
 211   this._emitStatusIfDone();
 212 }
 213
 214 ClientReadableStream.prototype._receiveStatus = _receiveStatus;
 215
 216 /**
 217  * If we have both processed all incoming messages and received the status from
 218  * the server, emit the status. Otherwise, do nothing.
 219  * @private
 220  */
 221 function _emitStatusIfDone() {
 222   /* jshint validthis: true */
 223   var status;
 224   if (this.read_status && this.received_status) {
 225     if (this.read_status.code !== constants.status.OK) {
 226       status = this.read_status;
 227     } else {
 228       status = this.received_status;
 229     }
 230     if (status.code === constants.status.OK) {
 231       this.push(null);
 232     } else {
 233       var error = common.createStatusError(status);
 234       this.emit('error', error);
 235     }
 236     this.emit('status', status);
 237   }
 238 }
 239
 240 ClientReadableStream.prototype._emitStatusIfDone = _emitStatusIfDone;
 241
 242 /**
 243  * Read the next object from the stream.
 244  * @private
 245  * @param {*} size Ignored because we use objectMode=true
 246  */
 247 function _read(size) {
 248   /* jshint validthis: true */
 249   if (this.finished) {
 250     this.push(null);
 251   } else {
 252     if (!this.reading) {
 253       this.reading = true;
 254       var context = {
 255         stream: this
 256       };
 257       this.call.recvMessageWithContext(context);
 258     }
 259   }
 260 }
 261
 262 ClientReadableStream.prototype._read = _read;
 263
 264 util.inherits(ClientDuplexStream, Duplex);
 265
 266 /**
 267  * A stream that the client can read from or write to. Used for calls with
 268  * duplex streaming.
 269  * @constructor grpc~ClientDuplexStream
 270  * @extends external:Duplex
 271  * @borrows grpc~ClientUnaryCall#cancel as grpc~ClientDuplexStream#cancel
 272  * @borrows grpc~ClientUnaryCall#getPeer as grpc~ClientDuplexStream#getPeer
 273  * @borrows grpc~ClientWritableStream#write as grpc~ClientDuplexStream#write
 274  * @borrows grpc~ClientUnaryCall#event:metadata as
 275  *     grpc~ClientDuplexStream#metadata
 276  * @borrows grpc~ClientUnaryCall#event:status as
 277  *     grpc~ClientDuplexStream#status
 278  * @param {InterceptingCall} call Exposes gRPC request operations, processed by
 279  *     an interceptor stack.
 280  */
 281 function ClientDuplexStream(call) {
 282   Duplex.call(this, {objectMode: true});
 283   this.call = call;
 284   /* Status generated from reading messages from the server. Overrides the
 285    * status from the server if not OK */
 286   this.read_status = null;
 287   /* Status received from the server. */
 288   this.received_status = null;
 289   var self = this;
 290   this.on('finish', function() {
 291     self.call.halfClose();
 292   });
 293 }
 294
 295 ClientDuplexStream.prototype._readsDone = _readsDone;
 296 ClientDuplexStream.prototype._receiveStatus = _receiveStatus;
 297 ClientDuplexStream.prototype._emitStatusIfDone = _emitStatusIfDone;
 298 ClientDuplexStream.prototype._read = _read;
 299 ClientDuplexStream.prototype._write = _write;
 300
 301 /**
 302  * Cancel the ongoing call. Results in the call ending with a CANCELLED status,
 303  * unless it has already ended with some other status.
 304  * @alias grpc~ClientUnaryCall#cancel
 305  */
 306 function cancel() {
 307   /* jshint validthis: true */
 308   this.call.cancel();
 309 }
 310
 311 ClientUnaryCall.prototype.cancel = cancel;
 312 ClientReadableStream.prototype.cancel = cancel;
 313 ClientWritableStream.prototype.cancel = cancel;
 314 ClientDuplexStream.prototype.cancel = cancel;
 315
 316 /**
 317  * Get the endpoint this call/stream is connected to.
 318  * @return {string} The URI of the endpoint
 319  * @alias grpc~ClientUnaryCall#getPeer
 320  */
 321 function getPeer() {
 322   /* jshint validthis: true */
 323   return this.call.getPeer();
 324 }
 325
 326 ClientUnaryCall.prototype.getPeer = getPeer;
 327 ClientReadableStream.prototype.getPeer = getPeer;
 328 ClientWritableStream.prototype.getPeer = getPeer;
 329 ClientDuplexStream.prototype.getPeer = getPeer;
 330
 331 /**
 332  * Any client call type
 333  * @typedef {(grpc~ClientUnaryCall|grpc~ClientReadableStream|
 334  *            grpc~ClientWritableStream|grpc~ClientDuplexStream)}
 335  *     grpc.Client~Call
 336  */
 337
 338 /**
 339  * Options that can be set on a call.
 340  * @typedef {Object} grpc.Client~CallOptions
 341  * @property {grpc~Deadline} deadline The deadline for the entire call to
 342  *     complete.
 343  * @property {string} host Server hostname to set on the call. Only meaningful
 344  *     if different from the server address used to construct the client.
 345  * @property {grpc.Client~Call} parent Parent call. Used in servers when
 346  *     making a call as part of the process of handling a call. Used to
 347  *     propagate some information automatically, as specified by
 348  *     propagate_flags.
 349  * @property {number} propagate_flags Indicates which properties of a parent
 350  *     call should propagate to this call. Bitwise combination of flags in
 351  *     {@link grpc.propagate}.
 352  * @property {grpc.credentials~CallCredentials} credentials The credentials that
 353  *     should be used to make this particular call.
 354  */
 355
 356 /**
 357  * Properties of a call, for use with a {@link grpc.Client~callInvocationTransformer}.
 358  * @typedef {Object} grpc.Client~CallProperties
 359  * @property {*} argument The call argument. Only preset if the method is unary or server streaming.
 360  * @property {grpc.Metadata} metadata The request metadata
 361  * @property {grpc~Call} call The call object that will be returned by the client method
 362  * @property {grpc.Channel} channel The channel that will be used to make a request
 363  * @property {grpc~MethodDefinition} methodDefinition The MethodDefinition object that describes this method
 364  * @property {grpc.Client~CallOptions} options The call options passed when making this request
 365  * @property {grpc.Client~requestCallback} callback The callback that will handle the response.
 366  *     Only present if this method is unary or client streaming.
 367  */
 368
 369 /**
 370  * Call invocation transformer. Has access to the full call properties before a
 371  * call is processed and can modify most of those properties. Some modifications
 372  * will have no effect or may cause problems.
 373  * @name grpc.Client~callInvocationTransformer
 374  * @function
 375  * @param {grpc.Client~CallProperties} callProperties The original call properties
 376  * @return {grpc.Client~CallProperties} The modified call properties.
 377  */
 378
 379 /**
 380  * A function that functionally replaces the Channel constructor.
 381  * @name grpc.Client~channelFactory
 382  * @function
 383  * @param {string} target The address of the server to connect to
 384  * @param {grpc.ChannelCredentials} credentials Channel credentials to use when connecting
 385  * @param {grpc~ChannelOptions} options A map of channel options that will be passed to the core.
 386  *     The available options are listed in
 387  *     [this document]{@link https://grpc.github.io/grpc/core/group__grpc__arg__keys.html}.
 388  * @returns {grpc.Channel} This can either be an actual channel object, or an object with the
 389  *     same API.
 390  */
 391
 392 /**
 393  * A generic gRPC client. Primarily useful as a base class for generated clients
 394  * @memberof grpc
 395  * @constructor
 396  * @param {string} address Server address to connect to
 397  * @param {grpc.credentials~ChannelCredentials} credentials Credentials to use
 398  *     to connect to the server
 399  * @param {Object} options Options to apply to channel creation. Any of
 400  *     [the channel arguments]{@link https://grpc.github.io/grpc/core/group__grpc__arg__keys.html}
 401  *     can be used here in addition to specific client options.
 402  * @param {grpc~Interceptor[]} [options.interceptors] Interceptors to apply to each request
 403  * @param {grpc~InterceptorProvider[]} [options.interceptor_providers] Interceptor providers
 404  *     to apply interceptors to each request depending on the method definition. At most
 405  *     one of the interceptors and interceptor_providers options may be set.
 406  * @param {grpc.Client~callInvocationTransformer=} options.callInvocationTransformer
 407  * @param {grpc.Channel=} options.channelOverride Channel to use instead of constructing a new one.
 408  *     If set, the address, credentials, channel arguments options, and channelFactoryOverride
 409  *     option will all be ignored.
 410  * @param {grpc.Client~channelFactory} options.channelFactoryOverride Function to use instead of
 411  *     the Channel constructor when creating the Client's channel.
 412  */
 413 function Client(address, credentials, options) {
 414   var self = this;
 415   if (!options) {
 416     options = {};
 417   }
 418
 419   // Resolve interceptor options and assign interceptors to each method
 420   if (Array.isArray(options.interceptor_providers) && Array.isArray(options.interceptors)) {
 421     throw new client_interceptors.InterceptorConfigurationError(
 422       'Both interceptors and interceptor_providers were passed as options ' +
 423       'to the client constructor. Only one of these is allowed.');
 424   }
 425   self.$interceptors = options.interceptors || [];
 426   self.$interceptor_providers = options.interceptor_providers || [];
 427   if (self.$method_definitions) {
 428     Object.keys(self.$method_definitions).forEach(method_name => {
 429       const method_definition = self.$method_definitions[method_name];
 430       self[method_name].interceptors = client_interceptors
 431         .resolveInterceptorProviders(self.$interceptor_providers, method_definition)
 432         .concat(self.$interceptors);
 433     });
 434   }
 435
 436   this.$callInvocationTransformer = options.callInvocationTransformer;
 437
 438   let channelOverride = options.channelOverride;
 439   let channelFactoryOverride = options.channelFactoryOverride;
 440   // Exclude channel options which have already been consumed
 441   const ignoredKeys = [
 442     'interceptors', 'interceptor_providers', 'channelOverride',
 443     'channelFactoryOverride', 'callInvocationTransformer'
 444   ];
 445   var channel_options = Object.getOwnPropertyNames(options)
 446     .reduce((acc, key) => {
 447       if (ignoredKeys.indexOf(key) === -1) {
 448         acc[key] = options[key];
 449       }
 450       return acc;
 451     }, {});
 452   /* Private fields use $ as a prefix instead of _ because it is an invalid
 453    * prefix of a method name */
 454   if (channelOverride) {
 455     this.$channel = options.channelOverride;
 456   } else {
 457     if (channelFactoryOverride) {
 458       this.$channel = channelFactoryOverride(address, credentials, channel_options);
 459     } else {
 460       this.$channel = new grpc.Channel(address, credentials, channel_options);
 461     }
 462   }
 463 }
 464
 465 exports.Client = Client;
 466
 467 Client.prototype.resolveCallInterceptors = function(method_definition, interceptors, interceptor_providers) {
 468   if (Array.isArray(interceptors) && Array.isArray(interceptor_providers)) {
 469     throw new client_interceptors.InterceptorConfigurationError(
 470       'Both interceptors and interceptor_providers were passed as call ' +
 471       'options. Only one of these is allowed.');
 472   }
 473   if (Array.isArray(interceptors) || Array.isArray(interceptor_providers)) {
 474     interceptors = interceptors || [];
 475     interceptor_providers = interceptor_providers || [];
 476     return client_interceptors.resolveInterceptorProviders(interceptor_providers, method_definition).concat(interceptors);
 477   } else {
 478     return client_interceptors.resolveInterceptorProviders(this.$interceptor_providers, method_definition).concat(this.$interceptors);
 479   }
 480 }
 481
 482 /**
 483  * @callback grpc.Client~requestCallback
 484  * @param {?grpc~ServiceError} error The error, if the call
 485  *     failed
 486  * @param {*} value The response value, if the call succeeded
 487  */
 488
 489 /**
 490  * Make a unary request to the given method, using the given serialize
 491  * and deserialize functions, with the given argument.
 492  * @param {string} path The path of the method to request
 493  * @param {grpc~serialize} serialize The serialization function for
 494  *     inputs
 495  * @param {grpc~deserialize} deserialize The deserialization
 496  *     function for outputs
 497  * @param {*} argument The argument to the call. Should be serializable with
 498  *     serialize
 499  * @param {grpc.Metadata=} metadata Metadata to add to the call
 500  * @param {grpc.Client~CallOptions=} options Options map
 501  * @param {grpc.Client~requestCallback} callback The callback
 502  *     for when the response is received
 503  * @return {grpc~ClientUnaryCall} An event emitter for stream related events
 504  */
 505 Client.prototype.makeUnaryRequest = function(path, serialize, deserialize,
 506                                              argument, metadata, options,
 507                                              callback) {
 508   if (typeof options === 'function') {
 509     callback = options;
 510     if (metadata instanceof Metadata) {
 511       options = {};
 512     } else {
 513       options = metadata;
 514       metadata = new Metadata();
 515     }
 516   } else if (typeof metadata === 'function') {
 517     callback = metadata;
 518     metadata = new Metadata();
 519     options = {};
 520   }
 521   if (!metadata) {
 522     metadata = new Metadata();
 523   }
 524   if (!options) {
 525     options = {};
 526   }
 527   if (!((metadata instanceof Metadata) &&
 528         (options instanceof Object) &&
 529         (typeof callback === 'function'))) {
 530     throw new Error('Argument mismatch in makeUnaryRequest');
 531   }
 532
 533   var method_definition = options.method_definition = {
 534     path: path,
 535     requestStream: false,
 536     responseStream: false,
 537     requestSerialize: serialize,
 538     responseDeserialize: deserialize
 539   };
 540
 541   metadata = metadata.clone();
 542
 543   var callProperties = {
 544     argument: argument,
 545     metadata: metadata,
 546     call: new ClientUnaryCall(),
 547     channel: this.$channel,
 548     methodDefinition: method_definition,
 549     callOptions: options,
 550     callback: callback
 551   };
 552
 553   // Transform call properties if specified.
 554   if (this.$callInvocationTransformer) {
 555     callProperties = this.$callInvocationTransformer(callProperties);
 556   }
 557
 558   var callOptions = callProperties.callOptions;
 559   var methodDefinition = callProperties.methodDefinition;
 560
 561   var interceptors = Client.prototype.resolveCallInterceptors.call(
 562     this,
 563     methodDefinition,
 564     callOptions.interceptors,
 565     callOptions.interceptor_providers
 566   );
 567
 568   var intercepting_call = client_interceptors.getInterceptingCall(
 569     methodDefinition,
 570     callOptions,
 571     interceptors,
 572     callProperties.channel,
 573     callProperties.callback
 574   );
 575
 576   var emitter = callProperties.call;
 577   emitter.call = intercepting_call;
 578
 579   var last_listener = client_interceptors.getLastListener(
 580     methodDefinition,
 581     emitter,
 582     callProperties.callback
 583   );
 584
 585   intercepting_call.start(callProperties.metadata, last_listener);
 586   intercepting_call.sendMessage(callProperties.argument);
 587   intercepting_call.halfClose();
 588
 589   return emitter;
 590 };
 591
 592 /**
 593  * Make a client stream request to the given method, using the given serialize
 594  * and deserialize functions, with the given argument.
 595  * @param {string} path The path of the method to request
 596  * @param {grpc~serialize} serialize The serialization function for
 597  *     inputs
 598  * @param {grpc~deserialize} deserialize The deserialization
 599  *     function for outputs
 600  * @param {grpc.Metadata=} metadata Array of metadata key/value pairs to add to
 601  *     the call
 602  * @param {grpc.Client~CallOptions=} options Options map
 603  * @param {grpc.Client~requestCallback} callback The callback for when the
 604  *     response is received
 605  * @return {grpc~ClientWritableStream} An event emitter for stream related
 606  *     events
 607  */
 608 Client.prototype.makeClientStreamRequest = function(path, serialize,
 609                                                     deserialize, metadata,
 610                                                     options, callback) {
 611   if (typeof options === 'function') {
 612     callback = options;
 613     if (metadata instanceof Metadata) {
 614       options = {};
 615     } else {
 616       options = metadata;
 617       metadata = new Metadata();
 618     }
 619   } else if (typeof metadata === 'function') {
 620     callback = metadata;
 621     metadata = new Metadata();
 622     options = {};
 623   }
 624   if (!metadata) {
 625     metadata = new Metadata();
 626   }
 627   if (!options) {
 628     options = {};
 629   }
 630   if (!((metadata instanceof Metadata) &&
 631        (options instanceof Object) &&
 632        (typeof callback === 'function'))) {
 633     throw new Error('Argument mismatch in makeClientStreamRequest');
 634   }
 635
 636   var method_definition = options.method_definition = {
 637     path: path,
 638     requestStream: true,
 639     responseStream: false,
 640     requestSerialize: serialize,
 641     responseDeserialize: deserialize
 642   };
 643
 644   metadata = metadata.clone();
 645
 646   var callProperties = {
 647     metadata: metadata,
 648     call: new ClientWritableStream(),
 649     channel: this.$channel,
 650     methodDefinition: method_definition,
 651     callOptions: options,
 652     callback: callback
 653   };
 654
 655   // Transform call properties if specified.
 656   if (this.$callInvocationTransformer) {
 657     callProperties = this.$callInvocationTransformer(callProperties);
 658   }
 659
 660   var callOptions = callProperties.callOptions;
 661   var methodDefinition = callProperties.methodDefinition;
 662
 663   var interceptors = Client.prototype.resolveCallInterceptors.call(
 664     this,
 665     methodDefinition,
 666     callOptions.interceptors,
 667     callOptions.interceptor_providers
 668   );
 669
 670   var intercepting_call = client_interceptors.getInterceptingCall(
 671     methodDefinition,
 672     callOptions,
 673     interceptors,
 674     callProperties.channel,
 675     callProperties.callback
 676   );
 677
 678   var emitter = callProperties.call;
 679   emitter.call = intercepting_call;
 680
 681   var last_listener = client_interceptors.getLastListener(
 682     methodDefinition,
 683     emitter,
 684     callProperties.callback
 685   );
 686
 687   intercepting_call.start(callProperties.metadata, last_listener);
 688
 689   return emitter;
 690 };
 691
 692 /**
 693  * Make a server stream request to the given method, with the given serialize
 694  * and deserialize function, using the given argument
 695  * @param {string} path The path of the method to request
 696  * @param {grpc~serialize} serialize The serialization function for inputs
 697  * @param {grpc~deserialize} deserialize The deserialization
 698  *     function for outputs
 699  * @param {*} argument The argument to the call. Should be serializable with
 700  *     serialize
 701  * @param {grpc.Metadata=} metadata Array of metadata key/value pairs to add to
 702  *     the call
 703  * @param {grpc.Client~CallOptions=} options Options map
 704  * @return {grpc~ClientReadableStream} An event emitter for stream related
 705  *     events
 706  */
 707 Client.prototype.makeServerStreamRequest = function(path, serialize,
 708                                                     deserialize, argument,
 709                                                     metadata, options) {
 710   if (!(metadata instanceof Metadata)) {
 711     options = metadata;
 712     metadata = new Metadata();
 713   }
 714   if (!(options instanceof Object)) {
 715     options = {};
 716   }
 717   if (!((metadata instanceof Metadata) && (options instanceof Object))) {
 718     throw new Error('Argument mismatch in makeServerStreamRequest');
 719   }
 720
 721   var method_definition = options.method_definition = {
 722     path: path,
 723     requestStream: false,
 724     responseStream: true,
 725     requestSerialize: serialize,
 726     responseDeserialize: deserialize
 727   };
 728
 729   metadata = metadata.clone();
 730
 731   var callProperties = {
 732     argument: argument,
 733     metadata: metadata,
 734     call: new ClientReadableStream(),
 735     channel: this.$channel,
 736     methodDefinition: method_definition,
 737     callOptions: options,
 738   };
 739
 740   // Transform call properties if specified.
 741   if (this.$callInvocationTransformer) {
 742     callProperties = this.$callInvocationTransformer(callProperties);
 743   }
 744
 745   var callOptions = callProperties.callOptions;
 746   var methodDefinition = callProperties.methodDefinition;
 747
 748   var interceptors = Client.prototype.resolveCallInterceptors.call(
 749     this,
 750     methodDefinition,
 751     callOptions.interceptors,
 752     callOptions.interceptor_providers
 753   );
 754
 755   var emitter = callProperties.call;
 756   var intercepting_call = client_interceptors.getInterceptingCall(
 757     methodDefinition,
 758     callOptions,
 759     interceptors,
 760     callProperties.channel,
 761     emitter
 762   );
 763   emitter.call = intercepting_call;
 764   var last_listener = client_interceptors.getLastListener(
 765     methodDefinition,
 766     emitter
 767   );
 768
 769   intercepting_call.start(callProperties.metadata, last_listener);
 770   intercepting_call.sendMessage(callProperties.argument);
 771   intercepting_call.halfClose();
 772
 773   return emitter;
 774 };
 775
 776 /**
 777  * Make a bidirectional stream request with this method on the given channel.
 778  * @param {string} path The path of the method to request
 779  * @param {grpc~serialize} serialize The serialization function for inputs
 780  * @param {grpc~deserialize} deserialize The deserialization
 781  *     function for outputs
 782  * @param {grpc.Metadata=} metadata Array of metadata key/value
 783  *     pairs to add to the call
 784  * @param {grpc.Client~CallOptions=} options Options map
 785  * @return {grpc~ClientDuplexStream} An event emitter for stream related events
 786  */
 787 Client.prototype.makeBidiStreamRequest = function(path, serialize,
 788                                                   deserialize, metadata,
 789                                                   options) {
 790   if (!(metadata instanceof Metadata)) {
 791     options = metadata;
 792     metadata = new Metadata();
 793   }
 794   if (!(options instanceof Object)) {
 795     options = {};
 796   }
 797   if (!((metadata instanceof Metadata) && (options instanceof Object))) {
 798     throw new Error('Argument mismatch in makeBidiStreamRequest');
 799   }
 800
 801   var method_definition = options.method_definition = {
 802     path: path,
 803     requestStream: true,
 804     responseStream: true,
 805     requestSerialize: serialize,
 806     responseDeserialize: deserialize
 807   };
 808
 809   metadata = metadata.clone();
 810
 811   var callProperties = {
 812     metadata: metadata,
 813     call: new ClientDuplexStream(),
 814     channel: this.$channel,
 815     methodDefinition: method_definition,
 816     callOptions: options,
 817   };
 818
 819   // Transform call properties if specified.
 820   if (this.$callInvocationTransformer) {
 821     callProperties = this.$callInvocationTransformer(callProperties);
 822   }
 823
 824   var callOptions = callProperties.callOptions;
 825   var methodDefinition = callProperties.methodDefinition;
 826
 827   var interceptors = Client.prototype.resolveCallInterceptors.call(
 828     this,
 829     methodDefinition,
 830     callOptions.interceptors,
 831     callOptions.interceptor_providers
 832   );
 833
 834
 835   var emitter = callProperties.call;
 836   var intercepting_call = client_interceptors.getInterceptingCall(
 837     methodDefinition,
 838     callOptions,
 839     interceptors,
 840     callProperties.channel,
 841     emitter
 842   );
 843   emitter.call = intercepting_call;
 844   var last_listener = client_interceptors.getLastListener(
 845     methodDefinition,
 846     emitter
 847   );
 848
 849   intercepting_call.start(callProperties.metadata, last_listener);
 850
 851   return emitter;
 852 };
 853
 854 /**
 855  * Close this client.
 856  */
 857 Client.prototype.close = function() {
 858   this.$channel.close();
 859 };
 860
 861 /**
 862  * Return the underlying channel object for the specified client
 863  * @return {Channel} The channel
 864  */
 865 Client.prototype.getChannel = function() {
 866   return this.$channel;
 867 };
 868
 869 /**
 870  * Wait for the client to be ready. The callback will be called when the
 871  * client has successfully connected to the server, and it will be called
 872  * with an error if the attempt to connect to the server has unrecoverablly
 873  * failed or if the deadline expires. This function will make the channel
 874  * start connecting if it has not already done so.
 875  * @param {grpc~Deadline} deadline When to stop waiting for a connection.
 876  * @param {function(Error)} callback The callback to call when done attempting
 877  *     to connect.
 878  */
 879 Client.prototype.waitForReady = function(deadline, callback) {
 880   var self = this;
 881   var checkState = function(err) {
 882     if (err) {
 883       callback(new Error('Failed to connect before the deadline'));
 884       return;
 885     }
 886     var new_state;
 887     try {
 888       new_state = self.$channel.getConnectivityState(true);
 889     } catch (e) {
 890       callback(new Error('The channel has been closed'));
 891       return;
 892     }
 893     if (new_state === grpc.connectivityState.READY) {
 894       callback();
 895     } else if (new_state === grpc.connectivityState.FATAL_FAILURE) {
 896       callback(new Error('Failed to connect to server'));
 897     } else {
 898       try {
 899         self.$channel.watchConnectivityState(new_state, deadline, checkState);
 900       } catch (e) {
 901         callback(new Error('The channel has been closed'));
 902       }
 903     }
 904   };
 905   /* Force a single round of polling to ensure that the channel state is up
 906    * to date */
 907   grpc.forcePoll();
 908   setImmediate(checkState);
 909 };
 910
 911 /**
 912  * Map with short names for each of the requester maker functions. Used in
 913  * makeClientConstructor
 914  * @private
 915  */
 916 var requester_funcs = {
 917   [methodTypes.UNARY]: Client.prototype.makeUnaryRequest,
 918   [methodTypes.CLIENT_STREAMING]: Client.prototype.makeClientStreamRequest,
 919   [methodTypes.SERVER_STREAMING]: Client.prototype.makeServerStreamRequest,
 920   [methodTypes.BIDI_STREAMING]: Client.prototype.makeBidiStreamRequest
 921 };
 922
 923 function getDefaultValues(metadata, options) {
 924   var res = {};
 925   res.metadata = metadata || new Metadata();
 926   res.options = options || {};
 927   return res;
 928 }
 929
 930 /**
 931  * Map with wrappers for each type of requester function to make it use the old
 932  * argument order with optional arguments after the callback.
 933  * @access private
 934  */
 935 var deprecated_request_wrap = {
 936   [methodTypes.UNARY]: function(makeUnaryRequest) {
 937     return function makeWrappedUnaryRequest(argument, callback,
 938                                             metadata, options) {
 939       /* jshint validthis: true */
 940       var opt_args = getDefaultValues(metadata, options);
 941       return makeUnaryRequest.call(this, argument, opt_args.metadata,
 942                                    opt_args.options, callback);
 943     };
 944   },
 945   [methodTypes.CLIENT_STREAMING]: function(makeServerStreamRequest) {
 946     return function makeWrappedClientStreamRequest(callback, metadata,
 947                                                    options) {
 948       /* jshint validthis: true */
 949       var opt_args = getDefaultValues(metadata, options);
 950       return makeServerStreamRequest.call(this, opt_args.metadata,
 951                                           opt_args.options, callback);
 952     };
 953   },
 954   [methodTypes.SERVER_STREAMING]: x => x,
 955   [methodTypes.BIDI_STREAMING]: x => x
 956 };
 957
 958 /**
 959  * Creates a constructor for a client with the given methods, as specified in
 960  * the methods argument. The resulting class will have an instance method for
 961  * each method in the service, which is a partial application of one of the
 962  * [Client]{@link grpc.Client} request methods, depending on `requestSerialize`
 963  * and `responseSerialize`, with the `method`, `serialize`, and `deserialize`
 964  * arguments predefined.
 965  * @memberof grpc
 966  * @alias grpc~makeGenericClientConstructor
 967  * @param {grpc~ServiceDefinition} methods An object mapping method names to
 968  *     method attributes
 969  * @param {string} serviceName The fully qualified name of the service
 970  * @param {Object} class_options An options object.
 971  * @param {boolean=} [class_options.deprecatedArgumentOrder=false] Indicates
 972  *     that the old argument order should be used for methods, with optional
 973  *     arguments at the end instead of the callback at the end. This option
 974  *     is only a temporary stopgap measure to smooth an API breakage.
 975  *     It is deprecated, and new code should not use it.
 976  * @return {function} New client constructor, which is a subclass of
 977  *     {@link grpc.Client}, and has the same arguments as that constructor.
 978  */
 979 exports.makeClientConstructor = function(methods, serviceName,
 980                                          class_options) {
 981   if (!class_options) {
 982     class_options = {};
 983   }
 984
 985   function ServiceClient(address, credentials, options) {
 986     Client.call(this, address, credentials, options);
 987   }
 988
 989   util.inherits(ServiceClient, Client);
 990   ServiceClient.prototype.$method_definitions = methods;
 991   ServiceClient.prototype.$method_names = {};
 992
 993   Object.keys(methods).forEach(name => {
 994     const attrs = methods[name];
 995     if (common.isPrototypePolluted(name)) {
 996       return;
 997     }
 998     if (name.indexOf('$') === 0) {
 999       throw new Error('Method names cannot start with $');
1000     }
1001     var method_type = common.getMethodType(attrs);
1002     var method_func = function() {
1003       return requester_funcs[method_type].apply(this,
1004         [ attrs.path, attrs.requestSerialize, attrs.responseDeserialize ]
1005         .concat([].slice.call(arguments))
1006       );
1007     };
1008     if (class_options.deprecatedArgumentOrder) {
1009       ServiceClient.prototype[name] =
1010         deprecated_request_wrap[method_type](method_func);
1011     } else {
1012       ServiceClient.prototype[name] = method_func;
1013     }
1014     ServiceClient.prototype.$method_names[attrs.path] = name;
1015     // Associate all provided attributes with the method
1016     Object.assign(ServiceClient.prototype[name], attrs);
1017     if (attrs.originalName && !common.isPrototypePolluted(attrs.originalName)) {
1018       ServiceClient.prototype[attrs.originalName] =
1019         ServiceClient.prototype[name];
1020     }
1021   });
1022
1023   ServiceClient.service = methods;
1024
1025   return ServiceClient;
1026 };
1027
1028 /**
1029  * Return the underlying channel object for the specified client
1030  * @memberof grpc
1031  * @alias grpc~getClientChannel
1032  * @param {grpc.Client} client The client
1033  * @return {Channel} The channel
1034  * @see grpc.Client#getChannel
1035  */
1036 exports.getClientChannel = function(client) {
1037   return Client.prototype.getChannel.call(client);
1038 };
1039
1040 /**
1041  * Gets a map of client method names to interceptor stacks.
1042  * @param {grpc.Client} client
1043  * @returns {Object.<string, Interceptor[]>}
1044  */
1045 exports.getClientInterceptors = function(client) {
1046   return Object.keys(client.$method_definitions)
1047     .reduce((acc, key) => {
1048       if (typeof key === 'string') {
1049         acc[key] = client[key].interceptors;
1050       }
1051       return acc;
1052     }, {});
1053 };
1054
1055 /**
1056  * Wait for the client to be ready. The callback will be called when the
1057  * client has successfully connected to the server, and it will be called
1058  * with an error if the attempt to connect to the server has unrecoverablly
1059  * failed or if the deadline expires. This function will make the channel
1060  * start connecting if it has not already done so.
1061  * @memberof grpc
1062  * @alias grpc~waitForClientReady
1063  * @param {grpc.Client} client The client to wait on
1064  * @param {grpc~Deadline} deadline When to stop waiting for a connection. Pass
1065  *     Infinity to wait forever.
1066  * @param {function(Error)} callback The callback to call when done attempting
1067  *     to connect.
1068  * @see grpc.Client#waitForReady
1069  */
1070 exports.waitForClientReady = function(client, deadline, callback) {
1071   Client.prototype.waitForReady.call(client, deadline, callback);
1072 };
1073
1074 exports.StatusBuilder = client_interceptors.StatusBuilder;
1075 exports.ListenerBuilder = client_interceptors.ListenerBuilder;
1076 exports.RequesterBuilder = client_interceptors.RequesterBuilder;
1077 exports.InterceptingCall = client_interceptors.InterceptingCall;