Skip to content

Commit 663ca69

Browse files
kamilmysliwieckamilmysliwiec
committed
feature(@nestjs/microservices) more descriptive gRPC errors
1 parent 6ad5ae6 commit 663ca69

36 files changed

Lines changed: 323 additions & 252 deletions

‎bundle/common/pipes/validation.pipe.d.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@ export interface ValidationPipeOptions extends ValidatorOptions {
55
transform?: boolean;
66
}
77
export declare class ValidationPipe implements PipeTransform<any> {
8-
private isTransformEnabled;
9-
private validatorOptions;
8+
protected isTransformEnabled: boolean;
9+
protected validatorOptions: ValidatorOptions;
1010
constructor(options?: ValidationPipeOptions);
1111
transform(value: any, metadata: ArgumentMetadata): Promise<any>;
1212
private toValidate(metadata);

‎bundle/microservices/client/client-grpc.d.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,5 +16,6 @@ export declare class ClientGrpcProxy extends ClientProxy implements ClientGrpc {
1616
loadProto(): any;
1717
lookupPackage(root: any, packageName: string): any;
1818
close(): void;
19+
connect(): Promise<any>;
1920
protected publish(partialPacket: any, callback: (packet) => any): Promise<void>;
2021
}

‎bundle/microservices/client/client-grpc.js‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,9 @@ class ClientGrpcProxy extends client_proxy_1.ClientProxy {
9191
this.grpcClient && this.grpcClient.close();
9292
this.grpcClient = null;
9393
}
94+
async connect() {
95+
throw new Error('The "connect()" method is not supported in gRPC mode.');
96+
}
9497
async publish(partialPacket, callback) {
9598
throw new Error('Method is not supported in gRPC mode. Use ClientGrpc instead (learn more in the documentation).');
9699
}

‎bundle/microservices/client/client-mqtt.d.ts‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
/// <reference types="node" />
21
import { ClientProxy } from './client-proxy';
32
import { ClientOptions } from '../interfaces/client-metadata.interface';
43
import { WritePacket } from './../interfaces';
@@ -10,11 +9,11 @@ export declare class ClientMqtt extends ClientProxy {
109
private readonly url;
1110
private mqttClient;
1211
constructor(options: ClientOptions);
13-
protected publish(partialPacket: ReadPacket, callback: (packet: WritePacket) => any): (channel: string, buffer: Buffer) => any;
1412
getAckPatternName(pattern: string): string;
1513
getResPatternName(pattern: string): string;
1614
close(): void;
17-
init(callback: (...args) => any): void;
15+
connect(): Promise<any>;
1816
createClient(): MqttClient;
19-
handleError(client: MqttClient, callback: (...args) => any): void;
17+
handleError(client: MqttClient): void;
18+
protected publish(partialPacket: ReadPacket, callback: (packet: WritePacket) => any): Promise<any>;
2019
}

‎bundle/microservices/client/client-mqtt.js‎

Lines changed: 45 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ const client_proxy_1 = require("./client-proxy");
44
const logger_service_1 = require("@nestjs/common/services/logger.service");
55
const constants_1 = require("./../constants");
66
const load_package_util_1 = require("@nestjs/common/utils/load-package.util");
7+
const constants_2 = require("./constants");
78
let mqttPackage = {};
89
class ClientMqtt extends client_proxy_1.ClientProxy {
910
constructor(options) {
@@ -14,38 +15,6 @@ class ClientMqtt extends client_proxy_1.ClientProxy {
1415
this.getOptionsProp(this.options, 'url') || constants_1.MQTT_DEFAULT_URL;
1516
mqttPackage = load_package_util_1.loadPackage('mqtt', ClientMqtt.name);
1617
}
17-
publish(partialPacket, callback) {
18-
if (!this.mqttClient) {
19-
this.init(callback);
20-
}
21-
const packet = this.assignPacketId(partialPacket);
22-
const pattern = JSON.stringify(partialPacket.pattern);
23-
const responseChannel = this.getResPatternName(pattern);
24-
const responseCallback = (channel, buffer) => {
25-
const { err, response, isDisposed, id } = JSON.parse(buffer.toString());
26-
if (id !== packet.id) {
27-
return undefined;
28-
}
29-
if (isDisposed || err) {
30-
callback({
31-
err,
32-
response: null,
33-
isDisposed: true,
34-
});
35-
this.mqttClient.unsubscribe(channel);
36-
this.mqttClient.removeListener(constants_1.MESSAGE_EVENT, responseCallback);
37-
return;
38-
}
39-
callback({
40-
err,
41-
response,
42-
});
43-
};
44-
this.mqttClient.on(constants_1.MESSAGE_EVENT, responseCallback);
45-
this.mqttClient.subscribe(responseChannel);
46-
this.mqttClient.publish(this.getAckPatternName(pattern), JSON.stringify(packet));
47-
return responseCallback;
48-
}
4918
getAckPatternName(pattern) {
5019
return `${pattern}_ack`;
5120
}
@@ -56,26 +25,56 @@ class ClientMqtt extends client_proxy_1.ClientProxy {
5625
this.mqttClient && this.mqttClient.end();
5726
this.mqttClient = null;
5827
}
59-
init(callback) {
28+
connect() {
6029
this.mqttClient = this.createClient();
61-
this.handleError(this.mqttClient, callback);
30+
return new Promise((resolve, reject) => {
31+
this.handleError(this.mqttClient);
32+
this.connect$(this.mqttClient)
33+
.subscribe(resolve, reject);
34+
});
6235
}
6336
createClient() {
6437
return mqttPackage.connect(this.url, this.options.options);
6538
}
66-
handleError(client, callback) {
67-
const errorCallback = err => {
68-
if (err.code === 'ECONNREFUSED') {
69-
callback(err, null);
70-
this.mqttClient = null;
39+
handleError(client) {
40+
client.addListener(constants_1.ERROR_EVENT, err => err.code !== constants_2.ECONNREFUSED && this.logger.error(err));
41+
}
42+
async publish(partialPacket, callback) {
43+
try {
44+
if (!this.mqttClient) {
45+
await this.connect();
7146
}
72-
this.logger.error(err);
73-
};
74-
client.addListener(constants_1.ERROR_EVENT, errorCallback);
75-
client.on(constants_1.CONNECT_EVENT, () => {
76-
client.removeListener(constants_1.ERROR_EVENT, errorCallback);
77-
client.addListener(constants_1.ERROR_EVENT, err => this.logger.error(err));
78-
});
47+
const packet = this.assignPacketId(partialPacket);
48+
const pattern = JSON.stringify(partialPacket.pattern);
49+
const responseChannel = this.getResPatternName(pattern);
50+
const responseCallback = (channel, buffer) => {
51+
const { err, response, isDisposed, id } = JSON.parse(buffer.toString());
52+
if (id !== packet.id) {
53+
return undefined;
54+
}
55+
if (isDisposed || err) {
56+
callback({
57+
err,
58+
response: null,
59+
isDisposed: true,
60+
});
61+
this.mqttClient.unsubscribe(channel);
62+
this.mqttClient.removeListener(constants_1.MESSAGE_EVENT, responseCallback);
63+
return;
64+
}
65+
callback({
66+
err,
67+
response,
68+
});
69+
};
70+
this.mqttClient.on(constants_1.MESSAGE_EVENT, responseCallback);
71+
this.mqttClient.subscribe(responseChannel);
72+
this.mqttClient.publish(this.getAckPatternName(pattern), JSON.stringify(packet));
73+
return responseCallback;
74+
}
75+
catch (err) {
76+
callback({ err });
77+
}
7978
}
8079
}
8180
exports.ClientMqtt = ClientMqtt;

‎bundle/microservices/client/client-nats.d.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,18 @@
11
import { ClientProxy } from './client-proxy';
22
import { ClientOptions } from '../interfaces/client-metadata.interface';
3-
import { WritePacket, ReadPacket, PacketId } from './../interfaces';
3+
import { WritePacket, ReadPacket } from './../interfaces';
44
import { Client } from '../external/nats-client.interface';
55
export declare class ClientNats extends ClientProxy {
66
private readonly options;
77
private readonly logger;
88
private readonly url;
99
private natsClient;
1010
constructor(options: ClientOptions);
11-
protected publish(partialPacket: ReadPacket, callback: (packet: WritePacket) => any): Promise<(message: WritePacket<any> & PacketId) => void>;
1211
getAckPatternName(pattern: string): string;
1312
getResPatternName(pattern: string): string;
1413
close(): void;
15-
init(callback: (...args) => any): Promise<void>;
14+
connect(): Promise<any>;
1615
createClient(): Promise<Client>;
17-
handleError(client: Client, callback: (...args) => any): void;
16+
handleError(client: Client): void;
17+
protected publish(partialPacket: ReadPacket, callback: (packet: WritePacket) => any): Promise<any>;
1818
}

‎bundle/microservices/client/client-nats.js‎

Lines changed: 43 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ const client_proxy_1 = require("./client-proxy");
44
const logger_service_1 = require("@nestjs/common/services/logger.service");
55
const constants_1 = require("./../constants");
66
const load_package_util_1 = require("@nestjs/common/utils/load-package.util");
7+
const constants_2 = require("./constants");
78
let natsPackage = {};
89
class ClientNats extends client_proxy_1.ClientProxy {
910
constructor(options) {
@@ -14,35 +15,6 @@ class ClientNats extends client_proxy_1.ClientProxy {
1415
this.getOptionsProp(this.options, 'url') || constants_1.NATS_DEFAULT_URL;
1516
natsPackage = load_package_util_1.loadPackage('nats', ClientNats.name);
1617
}
17-
async publish(partialPacket, callback) {
18-
if (!this.natsClient) {
19-
await this.init(callback);
20-
}
21-
const packet = this.assignPacketId(partialPacket);
22-
const pattern = JSON.stringify(partialPacket.pattern);
23-
const responseChannel = this.getResPatternName(pattern);
24-
const subscriptionHandler = (message) => {
25-
if (message.id !== packet.id) {
26-
return undefined;
27-
}
28-
const { err, response, isDisposed } = message;
29-
if (isDisposed || err) {
30-
callback({
31-
err,
32-
response: null,
33-
isDisposed: true,
34-
});
35-
return this.natsClient.unsubscribe(subscriptionId);
36-
}
37-
callback({
38-
err,
39-
response,
40-
});
41-
};
42-
const subscriptionId = this.natsClient.subscribe(responseChannel, subscriptionHandler);
43-
this.natsClient.publish(this.getAckPatternName(pattern), packet);
44-
return subscriptionHandler;
45-
}
4618
getAckPatternName(pattern) {
4719
return `${pattern}_ack`;
4820
}
@@ -53,28 +25,54 @@ class ClientNats extends client_proxy_1.ClientProxy {
5325
this.natsClient && this.natsClient.close();
5426
this.natsClient = null;
5527
}
56-
async init(callback) {
28+
async connect() {
5729
this.natsClient = await this.createClient();
58-
this.handleError(this.natsClient, callback);
30+
return new Promise((resolve, reject) => {
31+
this.handleError(this.natsClient);
32+
this.connect$(this.natsClient)
33+
.subscribe(resolve, reject);
34+
});
5935
}
6036
createClient() {
6137
const options = this.options.options || {};
62-
const client = natsPackage.connect(Object.assign({}, options, { url: this.url, json: true }));
63-
return new Promise(resolve => client.on(constants_1.CONNECT_EVENT, resolve));
38+
return natsPackage.connect(Object.assign({}, options, { url: this.url, json: true }));
6439
}
65-
handleError(client, callback) {
66-
const errorCallback = err => {
67-
if (err.code === 'ECONNREFUSED') {
68-
callback(err, null);
69-
this.natsClient = null;
40+
handleError(client) {
41+
client.addListener(constants_1.ERROR_EVENT, err => err.code !== constants_2.CONN_ERR && this.logger.error(err));
42+
}
43+
async publish(partialPacket, callback) {
44+
try {
45+
if (!this.natsClient) {
46+
await this.connect();
7047
}
71-
this.logger.error(err);
72-
};
73-
client.addListener(constants_1.ERROR_EVENT, errorCallback);
74-
client.on(constants_1.CONNECT_EVENT, () => {
75-
client.removeListener(constants_1.ERROR_EVENT, errorCallback);
76-
client.addListener(constants_1.ERROR_EVENT, err => this.logger.error(err));
77-
});
48+
const packet = this.assignPacketId(partialPacket);
49+
const pattern = JSON.stringify(partialPacket.pattern);
50+
const responseChannel = this.getResPatternName(pattern);
51+
const subscriptionHandler = (message) => {
52+
if (message.id !== packet.id) {
53+
return undefined;
54+
}
55+
const { err, response, isDisposed } = message;
56+
if (isDisposed || err) {
57+
callback({
58+
err,
59+
response: null,
60+
isDisposed: true,
61+
});
62+
return this.natsClient.unsubscribe(subscriptionId);
63+
}
64+
callback({
65+
err,
66+
response,
67+
});
68+
};
69+
const subscriptionId = this.natsClient.subscribe(responseChannel, subscriptionHandler);
70+
this.natsClient.publish(this.getAckPatternName(pattern), packet);
71+
return subscriptionHandler;
72+
}
73+
catch (err) {
74+
callback({ err });
75+
}
7876
}
7977
}
8078
exports.ClientNats = ClientNats;

‎bundle/microservices/client/client-proxy.d.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,13 @@
11
import { Observable, Observer } from 'rxjs';
22
import { ReadPacket, PacketId, WritePacket, ClientOptions } from './../interfaces';
33
export declare abstract class ClientProxy {
4+
abstract connect(): Promise<any>;
45
abstract close(): any;
6+
send<TResult = any, TInput = any>(pattern: any, data: TInput): Observable<TResult>;
57
protected abstract publish(packet: ReadPacket, callback: (packet: WritePacket) => void): any;
6-
send<T = any>(pattern: any, data: any): Observable<T>;
78
protected createObserver<T>(observer: Observer<T>): (packet: WritePacket) => void;
89
protected assignPacketId(packet: ReadPacket): ReadPacket & PacketId;
10+
protected connect$(instance: any, errorEvent?: string, connectEvent?: string): Observable<any>;
911
protected getOptionsProp<T extends {
1012
options?;
1113
}>(obj: ClientOptions, prop: keyof T['options'], defaultValue?: any): any;

‎bundle/microservices/client/client-proxy.js‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,9 @@ Object.defineProperty(exports, "__esModule", { value: true });
33
const rxjs_1 = require("rxjs");
44
const shared_utils_1 = require("@nestjs/common/utils/shared.utils");
55
const invalid_message_exception_1 = require("../exceptions/invalid-message.exception");
6+
const rxjs_2 = require("rxjs");
7+
const operators_1 = require("rxjs/operators");
8+
const constants_1 = require("../constants");
69
class ClientProxy {
710
send(pattern, data) {
811
if (shared_utils_1.isNil(pattern) || shared_utils_1.isNil(data)) {
@@ -29,6 +32,13 @@ class ClientProxy {
2932
.substr(2, 5) + Date.now();
3033
return Object.assign(packet, { id });
3134
}
35+
connect$(instance, errorEvent = constants_1.ERROR_EVENT, connectEvent = constants_1.CONNECT_EVENT) {
36+
const error$ = rxjs_2.fromEvent(instance, errorEvent).pipe(operators_1.map(err => {
37+
throw err;
38+
}));
39+
const connect$ = rxjs_2.fromEvent(instance, connectEvent);
40+
return rxjs_2.merge(error$, connect$).pipe(operators_1.take(1));
41+
}
3242
getOptionsProp(obj, prop, defaultValue = undefined) {
3343
return obj && obj.options ? obj.options[prop] : defaultValue;
3444
}

‎bundle/microservices/client/client-redis.d.ts‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import { ClientProxy } from './client-proxy';
22
import { ClientOptions } from '../interfaces/client-metadata.interface';
33
import { WritePacket, ReadPacket } from './../interfaces';
44
import { RedisClient, ClientOpts, RetryStrategyOptions } from '../external/redis.interface';
5+
import { Subject } from 'rxjs';
56
export declare class ClientRedis extends ClientProxy {
67
private readonly options;
78
private readonly logger;
@@ -10,13 +11,13 @@ export declare class ClientRedis extends ClientProxy {
1011
private subClient;
1112
private isExplicitlyTerminated;
1213
constructor(options: ClientOptions);
13-
protected publish(partialPacket: ReadPacket, callback: (packet: WritePacket) => any): Promise<(channel: string, buffer: string) => any>;
1414
getAckPatternName(pattern: string): string;
1515
getResPatternName(pattern: string): string;
1616
close(): void;
17-
init(callback: (...args) => any): void;
18-
createClient(): RedisClient;
19-
handleError(client: RedisClient, callback: (...args) => any): void;
20-
getClientOptions(): Partial<ClientOpts>;
21-
createRetryStrategy(options: RetryStrategyOptions): undefined | number;
17+
connect(): Promise<any>;
18+
createClient(error$: Subject<Error>): RedisClient;
19+
handleError(client: RedisClient): void;
20+
getClientOptions(error$: Subject<Error>): Partial<ClientOpts>;
21+
createRetryStrategy(options: RetryStrategyOptions, error$: Subject<Error>): undefined | number | Error;
22+
protected publish(partialPacket: ReadPacket, callback: (packet: WritePacket) => any): Promise<(channel: string, buffer: string) => any>;
2223
}

0 commit comments

Comments
 (0)