Skip to content

Commit

Permalink
Merge pull request #8365 from sjkummer/fix-grpc-channel-options
Browse files Browse the repository at this point in the history
Fix grpc channel options
  • Loading branch information
kamilmysliwiec committed Nov 8, 2021
2 parents d47fc51 + fdf11a5 commit 4bf2fd4
Show file tree
Hide file tree
Showing 4 changed files with 37 additions and 53 deletions.
46 changes: 17 additions & 29 deletions packages/microservices/client/client-grpc.ts
Expand Up @@ -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 = {};
Expand Down Expand Up @@ -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,
};

Expand Down
2 changes: 0 additions & 2 deletions packages/microservices/constants.ts
Expand Up @@ -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';
Expand Down
4 changes: 4 additions & 0 deletions packages/microservices/external/grpc-options.interface.ts
Expand Up @@ -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;
Expand All @@ -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;
}
38 changes: 16 additions & 22 deletions packages/microservices/server/server-grpc.ts
Expand Up @@ -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';
Expand All @@ -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 = {};
Expand Down Expand Up @@ -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) => {
Expand Down

0 comments on commit 4bf2fd4

Please sign in to comment.