Name resolution load balancing

packages/grpc-js/package.json

@@ -20,8 +20,9 @@
     "@types/gulp-mocha": "0.0.32",
     "@types/lodash": "^4.14.108",
     "@types/mocha": "^5.2.6",
+    "@types/node": "^12.7.5",
     "@types/ncp": "^2.0.1",
-    "@types/node": "^12.0.2",
+    "@types/node": "^12.7.5",
     "@types/pify": "^3.0.2",
     "@types/semver": "^6.0.1",
     "clang-format": "^1.0.55",

packages/grpc-js/src/backoff-timeout.ts

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2019 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+const INITIAL_BACKOFF_MS = 1000;
+const BACKOFF_MULTIPLIER = 1.6;
+const MAX_BACKOFF_MS = 120000;
+const BACKOFF_JITTER = 0.2;
+
+/**
+ * Get a number uniformly at random in the range [min, max)
+ * @param min
+ * @param max
+ */
+function uniformRandom(min: number, max: number) {
+  return Math.random() * (max - min) + min;
+}
+
+export interface BackoffOptions {
+  initialDelay?: number;
+  multiplier?: number;
+  jitter?: number;
+  maxDelay?: number;
+}
+
+export class BackoffTimeout {
+  private initialDelay: number = INITIAL_BACKOFF_MS;
+  private multiplier: number = BACKOFF_MULTIPLIER;
+  private maxDelay: number = MAX_BACKOFF_MS;
+  private jitter: number = BACKOFF_JITTER;
+  private nextDelay: number;
+  private timerId: NodeJS.Timer;
+  private running = false;
+
+  constructor(private callback: () => void, options?: BackoffOptions) {
+    if (options) {
+      if (options.initialDelay) {
+        this.initialDelay = options.initialDelay;
+      }
+      if (options.multiplier) {
+        this.multiplier = options.multiplier;
+      }
+      if (options.jitter) {
+        this.jitter = options.jitter;
+      }
+      if (options.maxDelay) {
+        this.maxDelay = options.maxDelay;
+      }
+    }
+    this.nextDelay = this.initialDelay;
+    this.timerId = setTimeout(() => {}, 0);
+    clearTimeout(this.timerId);
+  }
+
+  /**
+   * Call the callback after the current amount of delay time
+   */
+  runOnce() {
+    this.running = true;
+    this.timerId = setTimeout(() => {
+      this.callback();
+      this.running = false;
+    }, this.nextDelay);
+    const nextBackoff = Math.min(
+      this.nextDelay * this.multiplier,
+      this.maxDelay
+    );
+    const jitterMagnitude = nextBackoff * this.jitter;
+    this.nextDelay =
+      nextBackoff + uniformRandom(-jitterMagnitude, jitterMagnitude);
+  }
+
+  /**
+   * Stop the timer. The callback will not be called until `runOnce` is called
+   * again.
+   */
+  stop() {
+    clearTimeout(this.timerId);
+    this.running = false;
+  }
+
+  /**
+   * Reset the delay time to its initial value.
+   */
+  reset() {
+    this.nextDelay = this.initialDelay;
+  }
+
+  isRunning() {
+    return this.running;
+  }
+}

packages/grpc-js/src/call-credentials-filter.ts

@@ -17,14 +17,14 @@
 
 import { CallCredentials } from './call-credentials';
 import { Call } from './call-stream';
-import { Http2Channel } from './channel';
+import { Channel } from './channel';
 import { BaseFilter, Filter, FilterFactory } from './filter';
 import { Metadata } from './metadata';
 
 export class CallCredentialsFilter extends BaseFilter implements Filter {
   private serviceUrl: string;
   constructor(
-    private readonly channel: Http2Channel,
+    private readonly channel: Channel,
     private readonly stream: Call
   ) {
     super();
@@ -44,9 +44,7 @@ export class CallCredentialsFilter extends BaseFilter implements Filter {
   }
 
   async sendMetadata(metadata: Promise<Metadata>): Promise<Metadata> {
-    const channelCredentials = this.channel.credentials._getCallCredentials();
-    const streamCredentials = this.stream.getCredentials();
-    const credentials = channelCredentials.compose(streamCredentials);
+    const credentials = this.stream.getCredentials();
     const credsMetadata = credentials.generateMetadata({
       service_url: this.serviceUrl,
     });
@@ -58,8 +56,7 @@ export class CallCredentialsFilter extends BaseFilter implements Filter {
 
 export class CallCredentialsFilterFactory
   implements FilterFactory<CallCredentialsFilter> {
-  private readonly channel: Http2Channel;
-  constructor(channel: Http2Channel) {
+  constructor(private readonly channel: Channel) {
     this.channel = channel;
   }
 

packages/grpc-js/src/call-credentials.ts

@@ -16,6 +16,7 @@
  */
 
 import { Metadata } from './metadata';
+import { Call } from '.';
 
 export interface CallMetadataOptions {
   service_url: string;
@@ -44,6 +45,14 @@ export abstract class CallCredentials {
    */
   abstract compose(callCredentials: CallCredentials): CallCredentials;
 
+  /**
+   * Check whether two call credentials objects are equal. Separate
+   * SingleCallCredentials with identical metadata generator functions are
+   * equal.
+   * @param other The other CallCredentials object to compare with.
+   */
+  abstract _equals(other: CallCredentials): boolean;
+
   /**
    * Creates a new CallCredentials object from a given function that generates
    * Metadata objects.
@@ -81,6 +90,19 @@ class ComposedCallCredentials extends CallCredentials {
   compose(other: CallCredentials): CallCredentials {
     return new ComposedCallCredentials(this.creds.concat([other]));
   }
+
+  _equals(other: CallCredentials): boolean {
+    if (this === other) {
+      return true;
+    }
+    if (other instanceof ComposedCallCredentials) {
+      return this.creds.every((value, index) =>
+        value._equals(other.creds[index])
+      );
+    } else {
+      return false;
+    }
+  }
 }
 
 class SingleCallCredentials extends CallCredentials {
@@ -103,6 +125,17 @@ class SingleCallCredentials extends CallCredentials {
   compose(other: CallCredentials): CallCredentials {
     return new ComposedCallCredentials([this, other]);
   }
+
+  _equals(other: CallCredentials): boolean {
+    if (this === other) {
+      return true;
+    }
+    if (other instanceof SingleCallCredentials) {
+      return this.metadataGenerator === other.metadataGenerator;
+    } else {
+      return false;
+    }
+  }
 }
 
 class EmptyCallCredentials extends CallCredentials {
@@ -113,4 +146,8 @@ class EmptyCallCredentials extends CallCredentials {
   compose(other: CallCredentials): CallCredentials {
     return other;
   }
+
+  _equals(other: CallCredentials): boolean {
+    return other instanceof EmptyCallCredentials;
+  }
 }

packages/grpc-js/src/call-stream.ts

@@ -19,14 +19,14 @@ import * as http2 from 'http2';
 import { Duplex } from 'stream';
 
 import { CallCredentials } from './call-credentials';
-import { Http2Channel } from './channel';
 import { Status } from './constants';
 import { EmitterAugmentation1 } from './events';
 import { Filter } from './filter';
 import { FilterStackFactory } from './filter-stack';
 import { Metadata } from './metadata';
 import { ObjectDuplex, WriteCallback } from './object-stream';
 import { StreamDecoder } from './stream-decoder';
+import { ChannelImplementation } from './channel';
 
 const {
   HTTP2_HEADER_STATUS,
@@ -83,7 +83,7 @@ export type Call = {
   ObjectDuplex<WriteObject, Buffer>;
 
 export class Http2CallStream extends Duplex implements Call {
-  credentials: CallCredentials = CallCredentials.createEmpty();
+  credentials: CallCredentials;
   filterStack: Filter;
   private http2Stream: http2.ClientHttp2Stream | null = null;
   private pendingRead = false;
@@ -114,12 +114,14 @@ export class Http2CallStream extends Duplex implements Call {
 
   constructor(
     private readonly methodName: string,
-    private readonly channel: Http2Channel,
+    private readonly channel: ChannelImplementation,
     private readonly options: CallStreamOptions,
-    filterStackFactory: FilterStackFactory
+    filterStackFactory: FilterStackFactory,
+    private readonly channelCallCredentials: CallCredentials
   ) {
     super({ objectMode: true });
     this.filterStack = filterStackFactory.createFilter(this);
+    this.credentials = channelCallCredentials;
   }
 
   /**
@@ -358,12 +360,7 @@ export class Http2CallStream extends Duplex implements Call {
   }
 
   sendMetadata(metadata: Metadata): void {
-    this.channel._startHttp2Stream(
-      this.options.host,
-      this.methodName,
-      this,
-      metadata
-    );
+    this.channel._startCallStream(this, metadata);
   }
 
   private destroyHttp2Stream() {
@@ -395,7 +392,7 @@ export class Http2CallStream extends Duplex implements Call {
   }
 
   setCredentials(credentials: CallCredentials): void {
-    this.credentials = credentials;
+    this.credentials = this.channelCallCredentials.compose(credentials);
   }
 
   getStatus(): StatusObject | null {