Skip to content

Commit f80cd9e

Browse files
refactor(@nestjs/microservices) extract magic strings into constants
1 parent adbfae2 commit f80cd9e

5 files changed

Lines changed: 19 additions & 14 deletions

File tree

packages/microservices/client/client-grpc.ts

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import { ClientOptions } from '../interfaces/client-metadata.interface';
88
import { GRPC_DEFAULT_URL } from './../constants';
99
import { ClientGrpc, GrpcOptions } from './../interfaces';
1010
import { ClientProxy } from './client-proxy';
11+
import { GRPC_CANCELLED } from './constants';
1112

1213
let grpcPackage: any = {};
1314

@@ -25,7 +26,7 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
2526
this.grpcClient = this.createClient();
2627
}
2728

28-
public getService<T extends {}>(name: keyof T): T {
29+
public getService<T extends {}>(name: string): T {
2930
const { options } = this.options as GrpcOptions;
3031
if (!this.grpcClient[name]) {
3132
throw new InvalidGrpcServiceException();
@@ -59,14 +60,15 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
5960
): (...args) => Observable<any> {
6061
return (...args) => {
6162
return new Observable(observer => {
62-
const call = client[methodName](...args);
6363
let isClientCanceled = false;
64+
const call = client[methodName](...args);
65+
6466
call.on('data', (data: any) => observer.next(data));
6567
call.on('error', (error: any) => {
66-
if (error.details === 'Cancelled') {
68+
if (error.details === GRPC_CANCELLED) {
6769
call.destroy();
68-
if ( isClientCanceled ) {
69-
return; // do not error if cancel was inititiated by Client
70+
if (isClientCanceled) {
71+
return;
7072
}
7173
}
7274
observer.error(error);
@@ -76,10 +78,11 @@ export class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
7678
observer.complete();
7779
});
7880
return () => {
79-
if (!call.finished) {
80-
isClientCanceled = true;
81-
call.cancel();
81+
if (call.finished) {
82+
return undefined;
8283
}
84+
isClientCanceled = true;
85+
call.cancel();
8386
};
8487
});
8588
};
Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,3 @@
11
export const ECONNREFUSED = 'ECONNREFUSED';
2-
export const CONN_ERR = 'CONN_ERR';
2+
export const CONN_ERR = 'CONN_ERR';
3+
export const GRPC_CANCELLED = 'Cancelled';

packages/microservices/constants.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ export const MESSAGE_EVENT = 'message';
1010
export const ERROR_EVENT = 'error';
1111
export const CLOSE_EVENT = 'close';
1212
export const SUBSCRIBE = 'subscribe';
13+
export const CANCEL_EVENT = 'cancelled';
1314

1415
export const PATTERN_METADATA = 'pattern';
1516
export const CLIENT_CONFIGURATION_METADATA = 'client';
Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
export interface ClientGrpc {
2-
getService<T extends {}>(name: keyof T): T;
2+
getService<T extends {}>(name: string): T;
33
}

packages/microservices/server/server-grpc.ts

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
1+
import { fromEvent } from 'rxjs';
2+
import { takeUntil } from 'rxjs/operators';
13
import { InvalidGrpcPackageException } from '../exceptions/invalid-grpc-package.exception';
24
import { InvalidProtoDefinitionException } from '../exceptions/invalid-proto-definition.exception';
35
import { GrpcOptions, MicroserviceOptions } from '../interfaces/microservice-configuration.interface';
4-
import { GRPC_DEFAULT_URL } from './../constants';
6+
import { CANCEL_EVENT, GRPC_DEFAULT_URL } from './../constants';
57
import { CustomTransportStrategy } from './../interfaces';
68
import { Server } from './server';
7-
import { fromEvent } from 'rxjs';
8-
import { takeUntil } from 'rxjs/operators';
99

1010
let grpcPackage: any = {};
1111

@@ -106,7 +106,7 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
106106
const handler = methodHandler(call.request, call.metadata);
107107
const result$ = this.transformToObservable(await handler);
108108
await result$.pipe(
109-
takeUntil(fromEvent(call, 'cancelled')),
109+
takeUntil(fromEvent(call, CANCEL_EVENT)),
110110
).forEach(data => call.write(data));
111111
call.end();
112112
};

0 commit comments

Comments
 (0)