Skip to content

Commit 91be484

Browse files
kamilmysliwieckamilmysliwiec
committed
refactor(microservices) partially fixed rabbitmq transport (+tests)
1 parent 8ec7c5b commit 91be484

6 files changed

Lines changed: 433 additions & 324 deletions

File tree

‎packages/common/interfaces/microservices/microservice-configuration.interface.ts‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import { Transport } from '../../enums/transport.enum';
22
import { MqttClientOptions } from '../external/mqtt-options.interface';
33
import { CustomTransportStrategy } from './custom-transport-strategy.interface';
4-
import { Options } from 'amqplib';
54

65
export type MicroserviceOptions =
76
| GrpcOptions
@@ -38,7 +37,7 @@ export interface GrpcOptions {
3837
oneofs?: boolean;
3938
json?: boolean;
4039
includeDirs?: string[];
41-
}
40+
};
4241
};
4342
}
4443

@@ -88,6 +87,6 @@ export interface RmqOptions {
8887
queue?: string;
8988
prefetchCount?: number;
9089
isGlobalPrefetchCount?: boolean;
91-
queueOptions?: Options.AssertQueue;
90+
queueOptions?: any;
9291
};
93-
}
92+
}

‎packages/microservices/client/client-rmq.ts‎

Lines changed: 157 additions & 116 deletions
Original file line numberDiff line numberDiff line change
@@ -1,133 +1,174 @@
1-
import { Connection, Options } from 'amqplib';
2-
import { ERROR_EVENT, RQM_DEFAULT_URL, RQM_DEFAULT_QUEUE, RQM_DEFAULT_PREFETCH_COUNT, RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT, RQM_DEFAULT_QUEUE_OPTIONS, CONNECT_EVENT, DISCONNECT_EVENT } from './../constants';
31
import { Logger } from '@nestjs/common/services/logger.service';
42
import { loadPackage } from '@nestjs/common/utils/load-package.util';
5-
import { ClientProxy } from './client-proxy';
3+
import { randomStringGenerator } from '@nestjs/common/utils/random-string-generator.util';
4+
import * as amqp from 'amqp-connection-manager';
5+
import { EventEmitter } from 'events';
6+
import { fromEvent, merge, Observable } from 'rxjs';
7+
import { first, map, share, switchMap } from 'rxjs/operators';
68
import { ClientOptions, RmqOptions } from '../interfaces';
9+
import {
10+
DISCONNECT_EVENT,
11+
ERROR_EVENT,
12+
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT,
13+
RQM_DEFAULT_PREFETCH_COUNT,
14+
RQM_DEFAULT_QUEUE,
15+
RQM_DEFAULT_QUEUE_OPTIONS,
16+
RQM_DEFAULT_URL,
17+
} from './../constants';
718
import { WritePacket } from './../interfaces';
8-
import { EventEmitter } from 'events';
9-
import * as amqp from 'amqp-connection-manager';
19+
import { ClientProxy } from './client-proxy';
1020

1121
let rqmPackage: any = {};
1222

1323
export class ClientRMQ extends ClientProxy {
14-
private readonly logger = new Logger(ClientProxy.name);
15-
private client: any = null;
16-
private channel: any = null;
17-
private urls: string[];
18-
private queue: string;
19-
private prefetchCount: number;
20-
private isGlobalPrefetchCount: boolean;
21-
private queueOptions: Options.AssertQueue;
22-
private replyQueue: string;
23-
private responseEmitter: EventEmitter;
24-
25-
constructor(
26-
private readonly options: ClientOptions['options']) {
27-
super();
28-
this.urls =
29-
this.getOptionsProp<RmqOptions>(this.options, 'urls') || [RQM_DEFAULT_URL];
30-
this.queue =
31-
this.getOptionsProp<RmqOptions>(this.options, 'queue') || RQM_DEFAULT_QUEUE;
32-
this.prefetchCount =
33-
this.getOptionsProp<RmqOptions>(this.options, 'prefetchCount') || RQM_DEFAULT_PREFETCH_COUNT;
34-
this.isGlobalPrefetchCount =
35-
this.getOptionsProp<RmqOptions>(this.options, 'isGlobalPrefetchCount') || RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
36-
this.queueOptions =
37-
this.getOptionsProp<RmqOptions>(this.options, 'queueOptions') || RQM_DEFAULT_QUEUE_OPTIONS;
38-
rqmPackage = loadPackage('amqplib', ClientRMQ.name);
39-
this.connect();
40-
}
24+
protected readonly logger = new Logger(ClientProxy.name);
25+
protected connection: Promise<any>;
26+
protected client: any = null;
27+
protected channel: any = null;
28+
protected urls: string[];
29+
protected queue: string;
30+
protected prefetchCount: number;
31+
protected isGlobalPrefetchCount: boolean;
32+
protected queueOptions: any;
33+
protected replyQueue: string;
34+
protected responseEmitter: EventEmitter;
4135

42-
public close(): void {
43-
this.channel && this.channel.close();
44-
this.client && this.client.close();
45-
}
36+
constructor(protected readonly options: ClientOptions['options']) {
37+
super();
38+
this.urls = this.getOptionsProp<RmqOptions>(this.options, 'urls') || [
39+
RQM_DEFAULT_URL,
40+
];
41+
this.queue =
42+
this.getOptionsProp<RmqOptions>(this.options, 'queue') ||
43+
RQM_DEFAULT_QUEUE;
44+
this.prefetchCount =
45+
this.getOptionsProp<RmqOptions>(this.options, 'prefetchCount') ||
46+
RQM_DEFAULT_PREFETCH_COUNT;
47+
this.isGlobalPrefetchCount =
48+
this.getOptionsProp<RmqOptions>(this.options, 'isGlobalPrefetchCount') ||
49+
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
50+
this.queueOptions =
51+
this.getOptionsProp<RmqOptions>(this.options, 'queueOptions') ||
52+
RQM_DEFAULT_QUEUE_OPTIONS;
4653

47-
public listen() {
48-
this.channel.addSetup((channel) => {
49-
return Promise.all([
50-
channel.consume(this.replyQueue, (msg) => {
51-
this.responseEmitter.emit(msg.properties.correlationId, msg);
52-
}, { noAck: true }),
53-
]);
54-
});
55-
}
54+
rqmPackage = loadPackage('amqplib', ClientRMQ.name);
55+
}
5656

57-
public connect(): Promise<any> {
58-
if (this.client && this.channel) {
59-
return Promise.resolve();
60-
}
61-
return new Promise(async (resolve, reject) => {
62-
this.client = amqp.connect(this.urls);
63-
this.client.on(CONNECT_EVENT, x => {
64-
this.channel = this.client.createChannel({
65-
json: false,
66-
setup: async (channel) => {
67-
await channel.assertQueue(this.queue, this.queueOptions);
68-
await channel.prefetch(this.prefetchCount, this.isGlobalPrefetchCount);
69-
this.replyQueue = (await channel.assertQueue('', { exclusive: true })).queue;
70-
this.responseEmitter = new EventEmitter();
71-
this.responseEmitter.setMaxListeners(0);
72-
this.listen();
73-
resolve();
74-
},
75-
});
76-
});
77-
this.client.on(DISCONNECT_EVENT, err => {
78-
reject(err);
79-
this.client.close();
80-
this.client = null;
81-
});
82-
});
83-
}
57+
public close(): void {
58+
this.channel && this.channel.close();
59+
this.client && this.client.close();
60+
}
61+
62+
public listen() {
63+
this.channel.addSetup(channel =>
64+
channel.consume(
65+
this.replyQueue,
66+
msg => {
67+
this.responseEmitter.emit(msg.properties.correlationId, msg);
68+
},
69+
{ noAck: true },
70+
),
71+
);
72+
}
8473

85-
protected publish(messageObj, callback: (packet: WritePacket) => any) {
86-
if (!this.client) {
87-
this.connect().then(x => {
88-
this.sendMessage(messageObj, callback);
89-
});
90-
} else {
91-
this.sendMessage(messageObj, callback);
92-
}
74+
public connect(): Promise<any> {
75+
if (this.client && this.channel) {
76+
return this.connection;
9377
}
78+
this.client = this.createClient();
79+
this.handleError(this.client);
80+
81+
const connect$ = this.connect$(this.client);
82+
this.connection = this.mergeDisconnectEvent(this.client, connect$)
83+
.pipe(
84+
switchMap(() => {
85+
return new Promise(resolve =>
86+
this.client.createChannel({
87+
json: false,
88+
setup: async channel => this.setupChannel(channel, resolve),
89+
}),
90+
);
91+
}),
92+
share(),
93+
)
94+
.toPromise();
95+
return this.connection;
96+
}
97+
98+
public createClient<T = any>(): T {
99+
return amqp.connect(this.urls) as T;
100+
}
101+
102+
public mergeDisconnectEvent<T = any>(
103+
instance: any,
104+
source$: Observable<T>,
105+
): Observable<T> {
106+
const close$ = fromEvent(instance, DISCONNECT_EVENT).pipe(
107+
map(err => {
108+
throw err;
109+
}),
110+
);
111+
return merge(source$, close$).pipe(first());
112+
}
94113

95-
private sendMessage(messageObj, callback: (packet: WritePacket) => any) {
96-
try {
97-
const correlationId = Math.random().toString(36).substring(2, 15) + Math.random().toString(36).substring(2, 15);
98-
this.responseEmitter.on(correlationId, msg => {
99-
const { content } = msg;
100-
this.handleMessage(content, callback);
101-
});
102-
this.channel.sendToQueue(this.queue, Buffer.from(JSON.stringify(messageObj)), {
103-
replyTo: this.replyQueue,
104-
correlationId,
105-
});
106-
} catch (err) {
107-
this.logger.error(err);
108-
callback({ err });
109-
}
114+
public async setupChannel(channel: any, resolve: Function) {
115+
await channel.assertQueue(this.queue, this.queueOptions);
116+
await channel.prefetch(this.prefetchCount, this.isGlobalPrefetchCount);
117+
this.replyQueue = (await channel.assertQueue('', {
118+
exclusive: true,
119+
})).queue;
120+
121+
this.responseEmitter = new EventEmitter();
122+
this.responseEmitter.setMaxListeners(0);
123+
this.listen();
124+
125+
resolve();
126+
}
127+
128+
protected publish(
129+
message: any,
130+
callback: (packet: WritePacket) => any,
131+
): Function {
132+
try {
133+
const correlationId = randomStringGenerator();
134+
const listener = msg => {
135+
const { content } = msg;
136+
this.handleMessage(content, callback);
137+
};
138+
this.responseEmitter.on(correlationId, listener);
139+
this.channel.sendToQueue(
140+
this.queue,
141+
Buffer.from(JSON.stringify(message)),
142+
{
143+
replyTo: this.replyQueue,
144+
correlationId,
145+
},
146+
);
147+
return () => this.responseEmitter.removeListener(correlationId, listener);
148+
} catch (err) {
149+
callback({ err });
110150
}
151+
}
111152

112-
public handleMessage(
113-
msg: WritePacket,
114-
callback: (packet: WritePacket) => any,
115-
) {
116-
const { err, response, isDisposed } = JSON.parse(msg.toString());
117-
if (isDisposed || err) {
118-
callback({
119-
err,
120-
response: null,
121-
isDisposed: true,
122-
});
123-
}
124-
callback({
125-
err,
126-
response,
127-
});
128-
}
129-
130-
public handleError(client: Connection): void {
131-
client.addListener(ERROR_EVENT, err => this.logger.error(err));
153+
public handleMessage(
154+
packet: WritePacket,
155+
callback: (packet: WritePacket) => any,
156+
) {
157+
const { err, response, isDisposed } = packet;
158+
if (isDisposed || err) {
159+
callback({
160+
err,
161+
response: null,
162+
isDisposed: true,
163+
});
132164
}
133-
}
165+
callback({
166+
err,
167+
response,
168+
});
169+
}
170+
171+
public handleError(client: any): void {
172+
client.addListener(ERROR_EVENT, err => this.logger.error(err));
173+
}
174+
}

‎packages/microservices/client/constants.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
export const ECONNREFUSED = 'ECONNREFUSED';
22
export const CONN_ERR = 'CONN_ERR';
3-
export const GRPC_CANCELLED = 'Cancelled';
3+
export const GRPC_CANCELLED = 'Cancelled';

‎packages/microservices/interfaces/microservice-configuration.interface.ts‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import { MqttClientOptions } from '@nestjs/common/interfaces/external/mqtt-options.interface';
22
import { Transport } from '../enums/transport.enum';
33
import { Server } from './../server/server';
4-
import { Options } from 'amqplib';
54
import { CustomTransportStrategy } from './custom-transport-strategy.interface';
65

76
export type MicroserviceOptions =
@@ -39,7 +38,7 @@ export interface GrpcOptions {
3938
oneofs?: boolean;
4039
json?: boolean;
4140
includeDirs?: string[];
42-
}
41+
};
4342
};
4443
}
4544

@@ -89,6 +88,6 @@ export interface RmqOptions {
8988
queue?: string;
9089
prefetchCount?: number;
9190
isGlobalPrefetchCount?: boolean;
92-
queueOptions?: Options.AssertQueue;
91+
queueOptions?: any;
9392
};
94-
}
93+
}

0 commit comments

Comments
 (0)