Merge pull request #1200 from murgatroid99/grpc-js_channel_args

grpc-js: Add more channel args

PACKAGE-COMPARISON.md

@@ -3,7 +3,7 @@
 Feature | `grpc` | `@grpc/grpc-js`
 --------|--------|----------
 Client | :heavy_check_mark: | :heavy_check_mark:
-Server | :heavy_check_mark: | :x:
+Server | :heavy_check_mark: | :heavy_check_mark:
 Unary RPCs | :heavy_check_mark: | :heavy_check_mark:
 Streaming RPCs | :heavy_check_mark: | :heavy_check_mark:
 Deadlines | :heavy_check_mark: | :heavy_check_mark:
@@ -17,8 +17,8 @@ Connection Keepalives | :heavy_check_mark: | :heavy_check_mark:
 HTTP Connect Support | :heavy_check_mark: | :x:
 Retries | :heavy_check_mark: | :x:
 Stats/tracing/monitoring | :heavy_check_mark: | :x:
-Load Balancing | :heavy_check_mark: | :x:
+Load Balancing | :heavy_check_mark: | Pick first and round robin
-Initial Metadata Options | :heavy_check_mark: | :x:
+Initial Metadata Options | :heavy_check_mark: | only `waitForReady`
 
 Other Properties | `grpc` | `@grpc/grpc-js`
 -----------------|--------|----------------
@@ -37,5 +37,9 @@ In addition, all channel arguments defined in [this header file](https://github.
  - `grpc.keepalive_time_ms`
  - `grpc.keepalive_timeout_ms`
  - `grpc.service_config`
+ - `grpc.max_concurrent_streams`
+ - `grpc.initial_reconnect_backoff_ms`
+ - `grpc.max_reconnect_backoff_ms`
+ - `grpc.use_local_subchannel_pool`
  - `channelOverride`
  - `channelFactoryOverride`

packages/grpc-js/src/channel-options.ts

@@ -26,6 +26,10 @@ export interface ChannelOptions {
   'grpc.keepalive_time_ms'?: number;
   'grpc.keepalive_timeout_ms'?: number;
   'grpc.service_config'?: string;
+  'grpc.max_concurrent_streams'?: number;
+  'grpc.initial_reconnect_backoff_ms'?: number;
+  'grpc.max_reconnect_backoff_ms'?: number;
+  'grpc.use_local_subchannel_pool'?: number;
   [key: string]: string | number | undefined;
 }
 
@@ -41,6 +45,10 @@ export const recognizedOptions = {
   'grpc.keepalive_time_ms': true,
   'grpc.keepalive_timeout_ms': true,
   'grpc.service_config': true,
+  'grpc.max_concurrent_streams': true,
+  'grpc.initial_reconnect_backoff_ms': true,
+  'grpc.max_reconnect_backoff_ms': true,
+  'grpc.use_local_subchannel_pool': true,
 };
 
 export function channelOptionsEqual(

packages/grpc-js/src/channel.ts

@@ -129,8 +129,9 @@ export class ChannelImplementation implements Channel {
     private readonly credentials: ChannelCredentials,
     private readonly options: ChannelOptions
   ) {
-    // TODO(murgatroid99): check channel arg for getting a private pool
+    /* The global boolean parameter to getSubchannelPool has the inverse meaning to what
-    this.subchannelPool = getSubchannelPool(true);
+     * the grpc.use_local_subchannel_pool channel option means. */
+    this.subchannelPool = getSubchannelPool((options['grpc.use_local_subchannel_pool'] ?? 0) === 0);
     const channelControlHelper: ChannelControlHelper = {
       createSubchannel: (
         subchannelAddress: string,

packages/grpc-js/src/server.ts

@@ -45,6 +45,7 @@ import {
   ServerStatusResponse,
 } from './server-call';
 import { ServerCredentials } from './server-credentials';
+import { ChannelOptions } from './channel-options';
 
 function noop(): void {}
 
@@ -95,8 +96,11 @@ export class Server {
   >();
   private sessions = new Set<http2.ServerHttp2Session>();
   private started = false;
+  private options: ChannelOptions;
 
-  constructor(options?: object) {}
+  constructor(options?: ChannelOptions) {
+    this.options = options ?? {};
+  }
 
   addProtoService(): void {
     throw new Error('Not implemented. Use addService() instead');
@@ -197,13 +201,16 @@ export class Server {
 
     const url = new URL(`http://${port}`);
     const options: ListenOptions = { host: url.hostname, port: +url.port };
+    const serverOptions: http2.ServerOptions = {};
+    if ('grpc.max_concurrent_streams' in this.options) {
+      serverOptions.settings = {maxConcurrentStreams: this.options['grpc.max_concurrent_streams']};
+    }
 
     if (creds._isSecure()) {
-      this.http2Server = http2.createSecureServer(
+      const secureServerOptions = Object.assign(serverOptions, creds._getSettings()!);
-        creds._getSettings() as http2.SecureServerOptions
+      this.http2Server = http2.createSecureServer(secureServerOptions);
-      );
     } else {
-      this.http2Server = http2.createServer();
+      this.http2Server = http2.createServer(serverOptions);
     }
 
     this.http2Server.setTimeout(0, noop);

packages/grpc-js/src/subchannel.ts

@@ -22,7 +22,7 @@ import { Http2CallStream } from './call-stream';
 import { ChannelOptions } from './channel-options';
 import { PeerCertificate, checkServerIdentity } from 'tls';
 import { ConnectivityState } from './channel';
-import { BackoffTimeout } from './backoff-timeout';
+import { BackoffTimeout, BackoffOptions } from './backoff-timeout';
 import { getDefaultAuthority } from './resolver';
 import * as logging from './logging';
 import { LogVerbosity } from './constants';
@@ -170,6 +170,10 @@ export class Subchannel {
     clearTimeout(this.keepaliveIntervalId);
     this.keepaliveTimeoutId = setTimeout(() => {}, 0);
     clearTimeout(this.keepaliveTimeoutId);
+    const backoffOptions: BackoffOptions = {
+      initialDelay: options['grpc.initial_reconnect_backoff_ms'],
+      maxDelay: options['grpc.max_reconnect_backoff_ms']
+    };
     this.backoffTimeout = new BackoffTimeout(() => {
       if (this.continueConnecting) {
         this.transitionToState(
@@ -182,7 +186,7 @@ export class Subchannel {
           ConnectivityState.IDLE
         );
       }
-    });
+    }, backoffOptions);
   }
 
   /**