Merge pull request #8365 from sjkummer/fix-grpc-channel-options

Fix grpc channel options

Author: kamilmysliwiec

packages/microservices/client/client-grpc.ts

@@ -2,18 +2,14 @@ import { Logger } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
 import { isFunction, isObject } from '@nestjs/common/utils/shared.utils';
 import { Observable, Subscription } from 'rxjs';
-import {
-  GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
-  GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH,
-  GRPC_DEFAULT_PROTO_LOADER,
-  GRPC_DEFAULT_URL,
-} from '../constants';
+import { GRPC_DEFAULT_PROTO_LOADER, GRPC_DEFAULT_URL } from '../constants';
 import { InvalidGrpcPackageException } from '../errors/invalid-grpc-package.exception';
 import { InvalidGrpcServiceException } from '../errors/invalid-grpc-service.exception';
 import { InvalidProtoDefinitionException } from '../errors/invalid-proto-definition.exception';
 import { ClientGrpc, GrpcOptions } from '../interfaces';
 import { ClientProxy } from './client-proxy';
 import { GRPC_CANCELLED } from './constants';
+import { ChannelOptions } from '../external/grpc-options.interface';
 
 let grpcPackage: any = {};
 let grpcProtoLoaderPackage: any = {};
@@ -66,33 +62,25 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
       throw new InvalidGrpcServiceException();
     }
 
-    const maxSendMessageLengthKey = 'grpc.max_send_message_length';
-    const maxReceiveMessageLengthKey = 'grpc.max_receive_message_length';
-    const maxMessageLengthOptions = {
-      [maxSendMessageLengthKey]: this.getOptionsProp(
-        this.options,
-        'maxSendMessageLength',
-        GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH,
-      ),
-      [maxReceiveMessageLengthKey]: this.getOptionsProp(
-        this.options,
-        'maxReceiveMessageLength',
-        GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
-      ),
-    };
-    const maxMetadataSize = this.getOptionsProp(
-      this.options,
-      'maxMetadataSize',
-      -1,
-    );
-    if (maxMetadataSize > 0) {
-      maxMessageLengthOptions['grpc.max_metadata_size'] = maxMetadataSize;
+    const channelOptions: ChannelOptions =
+      this.options && this.options.channelOptions
+        ? this.options.channelOptions
+        : {};
+    if (this.options && this.options.maxSendMessageLength) {
+      channelOptions['grpc.max_send_message_length'] =
+        this.options.maxSendMessageLength;
+    }
+    if (this.options && this.options.maxReceiveMessageLength) {
+      channelOptions['grpc.max_receive_message_length'] =
+        this.options.maxReceiveMessageLength;
+    }
+    if (this.options && this.options.maxMetadataSize) {
+      channelOptions['grpc.max_metadata_size'] = this.options.maxMetadataSize;
     }
 
     const keepaliveOptions = this.getKeepaliveOptions();
     const options: Record<string, string | number> = {
-      ...(this.options.channelOptions || {}),
-      ...maxMessageLengthOptions,
+      ...channelOptions,
       ...keepaliveOptions,
     };
 

packages/microservices/constants.ts

@@ -39,8 +39,6 @@ export const GRPC_DEFAULT_PROTO_LOADER = '@grpc/proto-loader';
 export const NO_MESSAGE_HANDLER = `There is no matching message handler defined in the remote service.`;
 export const NO_EVENT_HANDLER = `There is no matching event handler defined in the remote service.`;
 export const DISCONNECTED_RMQ_MESSAGE = `Disconnected from RMQ. Trying to reconnect.`;
-export const GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH = 4 * 1024 * 1024;
-export const GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH = 4 * 1024 * 1024;
 
 export const KAFKA_DEFAULT_CLIENT = 'nestjs-consumer';
 export const KAFKA_DEFAULT_GROUP = 'nestjs-group';

packages/microservices/external/grpc-options.interface.ts

@@ -3,6 +3,9 @@
  * This listing is incomplete. Full reference: https://grpc.github.io/grpc/core/group__grpc__arg__keys.html
  */
 export interface ChannelOptions {
+  'grpc.max_send_message_length'?: number;
+  'grpc.max_receive_message_length'?: number;
+  'grpc.max_metadata_size'?: number;
   'grpc.ssl_target_name_override'?: string;
   'grpc.primary_user_agent'?: string;
   'grpc.secondary_user_agent'?: string;
@@ -12,5 +15,6 @@ export interface ChannelOptions {
   'grpc.initial_reconnect_backoff_ms'?: number;
   'grpc.max_reconnect_backoff_ms'?: number;
   'grpc.use_local_subchannel_pool'?: number;
+  'grpc-node.max_session_memory'?: number;
   [key: string]: string | number | undefined;
 }

packages/microservices/server/server-grpc.ts

@@ -7,8 +7,6 @@ import { EMPTY, fromEvent, lastValueFrom, Subject } from 'rxjs';
 import { catchError, takeUntil } from 'rxjs/operators';
 import {
   CANCEL_EVENT,
-  GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
-  GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH,
   GRPC_DEFAULT_PROTO_LOADER,
   GRPC_DEFAULT_URL,
 } from '../constants';
@@ -19,6 +17,7 @@ import { InvalidProtoDefinitionException } from '../errors/invalid-proto-definit
 import { CustomTransportStrategy, MessageHandler } from '../interfaces';
 import { GrpcOptions } from '../interfaces/microservice-configuration.interface';
 import { Server } from './server';
+import { ChannelOptions } from '../external/grpc-options.interface';
 
 let grpcPackage: any = {};
 let grpcProtoLoaderPackage: any = {};
@@ -333,27 +332,22 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
   }
 
   public async createClient(): Promise<any> {
-    const grpcOptions = {
-      'grpc.max_send_message_length': this.getOptionsProp(
-        this.options,
-        'maxSendMessageLength',
-        GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH,
-      ),
-      'grpc.max_receive_message_length': this.getOptionsProp(
-        this.options,
-        'maxReceiveMessageLength',
-        GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
-      ),
-    };
-    const maxMetadataSize = this.getOptionsProp(
-      this.options,
-      'maxMetadataSize',
-      -1,
-    );
-    if (maxMetadataSize > 0) {
-      grpcOptions['grpc.max_metadata_size'] = maxMetadataSize;
+    const channelOptions: ChannelOptions =
+      this.options && this.options.channelOptions
+        ? this.options.channelOptions
+        : {};
+    if (this.options && this.options.maxSendMessageLength) {
+      channelOptions['grpc.max_send_message_length'] =
+        this.options.maxSendMessageLength;
+    }
+    if (this.options && this.options.maxReceiveMessageLength) {
+      channelOptions['grpc.max_receive_message_length'] =
+        this.options.maxReceiveMessageLength;
+    }
+    if (this.options && this.options.maxMetadataSize) {
+      channelOptions['grpc.max_metadata_size'] = this.options.maxMetadataSize;
     }
-    const server = new grpcPackage.Server(grpcOptions);
+    const server = new grpcPackage.Server(channelOptions);
     const credentials = this.getOptionsProp(this.options, 'credentials');
 
     await new Promise((resolve, reject) => {