Resolved merge conflicts with master

1 files changed, 112 insertions, 51 deletions

diff --git a/src/node/src/client.js b/src/node/src/client.js
index de9efd7e4a..3c3642adeb 100644
--- a/src/node/src/client.js
+++ b/src/node/src/client.js
@@ -31,6 +31,11 @@
  *
  */
 
+/**
+ * Server module
+ * @module
+ */
+
 'use strict';
 
 var _ = require('lodash');
@@ -47,6 +52,7 @@ var Readable = stream.Readable;
 var Writable = stream.Writable;
 var Duplex = stream.Duplex;
 var util = require('util');
+var version = require('../package.json').version;
 
 util.inherits(ClientWritableStream, Writable);
 
@@ -71,6 +77,7 @@ function ClientWritableStream(call, serialize) {
 /**
  * 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
  * @param {function(Error=)} callback Called when the write is complete
@@ -111,6 +118,7 @@ function ClientReadableStream(call, deserialize) {
 
 /**
  * Read the next object from the stream.
+ * @access private
  * @param {*} size Ignored because we use objectMode=true
  */
 function _read(size) {
@@ -189,6 +197,43 @@ ClientWritableStream.prototype.cancel = cancel;
 ClientDuplexStream.prototype.cancel = cancel;
 
 /**
+ * Get the endpoint this call/stream is connected to.
+ * @return {string} The URI of the endpoint
+ */
+function getPeer() {
+  /* jshint validthis: true */
+  return this.call.getPeer();
+}
+
+ClientReadableStream.prototype.getPeer = getPeer;
+ClientWritableStream.prototype.getPeer = getPeer;
+ClientDuplexStream.prototype.getPeer = getPeer;
+
+/**
+ * Get a 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.
+ * @param {Object} options Options map.
+ */
+function getCall(channel, method, options) {
+  var deadline;
+  var host;
+  var parent;
+  var propagate_flags;
+  if (options) {
+    deadline = options.deadline;
+    host = options.host;
+    parent = _.get(options, 'parent.call');
+    propagate_flags = options.propagate_flags;
+  }
+  if (deadline === undefined) {
+    deadline = Infinity;
+  }
+  return new grpc.Call(channel, method, deadline, host,
+                       parent, propagate_flags);
+}
+
+/**
  * 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
@@ -207,25 +252,22 @@ function makeUnaryRequestFunction(method, serialize, deserialize) {
    *     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
-   * @param {number=} flags Flags for modifying how the message is sent.
-   *     Defaults to 0.
+   * @param {Object=} options Options map
    * @return {EventEmitter} An event emitter for stream related events
    */
-  function makeUnaryRequest(argument, callback, metadata, deadline, flags) {
+  function makeUnaryRequest(argument, callback, metadata, options) {
     /* jshint validthis: true */
-    if (deadline === undefined) {
-      deadline = Infinity;
-    }
     var emitter = new EventEmitter();
-    var call = new grpc.Call(this.channel, method, deadline);
+    var call = getCall(this.channel, method, options);
     if (metadata === null || metadata === undefined) {
       metadata = {};
     }
     emitter.cancel = function cancel() {
       call.cancel();
     };
+    emitter.getPeer = function getPeer() {
+      return call.getPeer();
+    };
     this.updateMetadata(this.auth_uri, metadata, function(error, metadata) {
       if (error) {
         call.cancel();
@@ -234,7 +276,7 @@ function makeUnaryRequestFunction(method, serialize, deserialize) {
       }
       var client_batch = {};
       var message = serialize(argument);
-      message.grpcWriteFlags = flags;
+      message.grpcWriteFlags = options.flags;
       client_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata;
       client_batch[grpc.opType.SEND_MESSAGE] = message;
       client_batch[grpc.opType.SEND_CLOSE_FROM_CLIENT] = true;
@@ -282,16 +324,12 @@ function makeClientStreamRequestFunction(method, serialize, deserialize) {
    *     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
+   * @param {Object=} options Options map
    * @return {EventEmitter} An event emitter for stream related events
    */
-  function makeClientStreamRequest(callback, metadata, deadline) {
+  function makeClientStreamRequest(callback, metadata, options) {
     /* jshint validthis: true */
-    if (deadline === undefined) {
-      deadline = Infinity;
-    }
-    var call = new grpc.Call(this.channel, method, deadline);
+    var call = getCall(this.channel, method, options);
     if (metadata === null || metadata === undefined) {
       metadata = {};
     }
@@ -356,18 +394,12 @@ function makeServerStreamRequestFunction(method, serialize, deserialize) {
    *     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
-   * @param {number=} flags Flags for modifying how the message is sent.
-   *     Defaults to 0.
+   * @param {Object} options Options map
    * @return {EventEmitter} An event emitter for stream related events
    */
-  function makeServerStreamRequest(argument, metadata, deadline, flags) {
+  function makeServerStreamRequest(argument, metadata, options) {
     /* jshint validthis: true */
-    if (deadline === undefined) {
-      deadline = Infinity;
-    }
-    var call = new grpc.Call(this.channel, method, deadline);
+    var call = getCall(this.channel, method, options);
     if (metadata === null || metadata === undefined) {
       metadata = {};
     }
@@ -380,7 +412,7 @@ function makeServerStreamRequestFunction(method, serialize, deserialize) {
       }
       var start_batch = {};
       var message = serialize(argument);
-      message.grpcWriteFlags = flags;
+      message.grpcWriteFlags = options.flags;
       start_batch[grpc.opType.SEND_INITIAL_METADATA] = metadata;
       start_batch[grpc.opType.RECV_INITIAL_METADATA] = true;
       start_batch[grpc.opType.SEND_MESSAGE] = message;
@@ -432,16 +464,12 @@ function makeBidiStreamRequestFunction(method, serialize, deserialize) {
    * @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
+   * @param {Options} options Options map
    * @return {EventEmitter} An event emitter for stream related events
    */
-  function makeBidiStreamRequest(metadata, deadline) {
+  function makeBidiStreamRequest(metadata, options) {
     /* jshint validthis: true */
-    if (deadline === undefined) {
-      deadline = Infinity;
-    }
-    var call = new grpc.Call(this.channel, method, deadline);
+    var call = getCall(this.channel, method, options);
     if (metadata === null || metadata === undefined) {
       metadata = {};
     }
@@ -509,31 +537,67 @@ var requester_makers = {
  * requestSerialize: function to serialize request objects
  * responseDeserialize: function to deserialize response objects
  * @param {Object} methods An object mapping method names to method attributes
- * @param {string} serviceName The name of the service
+ * @param {string} serviceName The fully qualified name of the service
  * @return {function(string, Object)} New client constructor
  */
-function makeClientConstructor(methods, serviceName) {
+exports.makeClientConstructor = function(methods, serviceName) {
   /**
    * Create a client with the given methods
    * @constructor
    * @param {string} address The address of the server to connect to
+   * @param {grpc.Credentials} credentials Credentials to use to connect
+   *     to the server
    * @param {Object} options Options to pass to the underlying channel
    * @param {function(string, Object, function)=} updateMetadata function to
    *     update the metadata for each request
    */
-  function Client(address, options, updateMetadata) {
+  function Client(address, credentials, options, updateMetadata) {
     if (!updateMetadata) {
       updateMetadata = function(uri, metadata, callback) {
         callback(null, metadata);
       };
     }
-
-    this.server_address = address.replace(/\/$/, '');
-    this.channel = new grpc.Channel(address, options);
-    this.auth_uri = this.server_address + '/' + serviceName;
+    if (!options) {
+      options = {};
+    }
+    options['grpc.primary_user_agent'] = 'grpc-node/' + version;
+    this.channel = new grpc.Channel(address, credentials, options);
+    // Remove the optional DNS scheme, trailing port, and trailing backslash
+    address = address.replace(/^(dns:\/{3})?([^:\/]+)(:\d+)?\/?$/, '$2');
+    this.server_address = address;
+    this.auth_uri = 'https://' + this.server_address + '/' + serviceName;
     this.updateMetadata = updateMetadata;
   }
 
+  /**
+   * Wait for the client to be ready. The callback will be called when the
+   * client has successfully connected to the server, and it will be called
+   * 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 {function(Error)} callback The callback to call when done attempting
+   *     to connect.
+   */
+  Client.prototype.$waitForReady = function(deadline, callback) {
+    var self = this;
+    var checkState = function(err) {
+      if (err) {
+        callback(new Error('Failed to connect before the deadline'));
+      }
+      var new_state = self.channel.getConnectivityState(true);
+      if (new_state === grpc.connectivityState.READY) {
+        callback();
+      } else if (new_state === grpc.connectivityState.FATAL_FAILURE) {
+        callback(new Error('Failed to connect to server'));
+      } else {
+        self.channel.watchConnectivityState(new_state, deadline, checkState);
+      }
+    };
+    checkState();
+  };
+
   _.each(methods, function(attrs, name) {
     var method_type;
     if (attrs.requestStream) {
@@ -558,7 +622,7 @@ function makeClientConstructor(methods, serviceName) {
   });
 
   return Client;
-}
+};
 
 /**
  * Creates a constructor for clients for the given service
@@ -566,22 +630,19 @@ function makeClientConstructor(methods, serviceName) {
  *     for
  * @return {function(string, Object)} New client constructor
  */
-function makeProtobufClientConstructor(service) {
+exports.makeProtobufClientConstructor =  function(service) {
   var method_attrs = common.getProtobufServiceAttrs(service, service.name);
-  var Client = makeClientConstructor(method_attrs);
+  var Client = exports.makeClientConstructor(
+      method_attrs, common.fullyQualifiedName(service));
   Client.service = service;
-
   return Client;
-}
-
-exports.makeClientConstructor = makeClientConstructor;
-
-exports.makeProtobufClientConstructor = makeProtobufClientConstructor;
+};
 
 /**
- * See docs for client.status
+ * Map of status code names to status codes
  */
 exports.status = grpc.status;
+
 /**
  * See docs for client.callError
  */