1 | "use strict";
|
2 | Object.defineProperty(exports, "__esModule", { value: true });
|
3 | exports.ClientRMQ = void 0;
|
4 | const logger_service_1 = require("@nestjs/common/services/logger.service");
|
5 | const load_package_util_1 = require("@nestjs/common/utils/load-package.util");
|
6 | const random_string_generator_util_1 = require("@nestjs/common/utils/random-string-generator.util");
|
7 | const shared_utils_1 = require("@nestjs/common/utils/shared.utils");
|
8 | const events_1 = require("events");
|
9 | const rxjs_1 = require("rxjs");
|
10 | const operators_1 = require("rxjs/operators");
|
11 | const constants_1 = require("../constants");
|
12 | const rmq_record_serializer_1 = require("../serializers/rmq-record.serializer");
|
13 | const client_proxy_1 = require("./client-proxy");
|
14 | let rqmPackage = {};
|
15 | const REPLY_QUEUE = 'amq.rabbitmq.reply-to';
|
16 |
|
17 |
|
18 |
|
19 | class ClientRMQ extends client_proxy_1.ClientProxy {
|
20 | constructor(options) {
|
21 | super();
|
22 | this.options = options;
|
23 | this.logger = new logger_service_1.Logger(client_proxy_1.ClientProxy.name);
|
24 | this.client = null;
|
25 | this.channel = null;
|
26 | this.urls = this.getOptionsProp(this.options, 'urls') || [constants_1.RQM_DEFAULT_URL];
|
27 | this.queue =
|
28 | this.getOptionsProp(this.options, 'queue') || constants_1.RQM_DEFAULT_QUEUE;
|
29 | this.queueOptions =
|
30 | this.getOptionsProp(this.options, 'queueOptions') ||
|
31 | constants_1.RQM_DEFAULT_QUEUE_OPTIONS;
|
32 | this.replyQueue =
|
33 | this.getOptionsProp(this.options, 'replyQueue') || REPLY_QUEUE;
|
34 | this.persistent =
|
35 | this.getOptionsProp(this.options, 'persistent') || constants_1.RQM_DEFAULT_PERSISTENT;
|
36 | this.noAssert =
|
37 | this.getOptionsProp(this.options, 'noAssert') || constants_1.RQM_DEFAULT_NO_ASSERT;
|
38 | (0, load_package_util_1.loadPackage)('amqplib', ClientRMQ.name, () => require('amqplib'));
|
39 | rqmPackage = (0, load_package_util_1.loadPackage)('amqp-connection-manager', ClientRMQ.name, () => require('amqp-connection-manager'));
|
40 | this.initializeSerializer(options);
|
41 | this.initializeDeserializer(options);
|
42 | }
|
43 | close() {
|
44 | this.channel && this.channel.close();
|
45 | this.client && this.client.close();
|
46 | this.channel = null;
|
47 | this.client = null;
|
48 | }
|
49 | connect() {
|
50 | if (this.client) {
|
51 | return this.convertConnectionToPromise();
|
52 | }
|
53 | this.client = this.createClient();
|
54 | this.handleError(this.client);
|
55 | this.handleDisconnectError(this.client);
|
56 | this.responseEmitter = new events_1.EventEmitter();
|
57 | this.responseEmitter.setMaxListeners(0);
|
58 | const connect$ = this.connect$(this.client);
|
59 | const withDisconnect$ = this.mergeDisconnectEvent(this.client, connect$).pipe((0, operators_1.switchMap)(() => this.createChannel()));
|
60 | const withReconnect$ = (0, rxjs_1.fromEvent)(this.client, constants_1.CONNECT_EVENT).pipe((0, operators_1.skip)(1));
|
61 | const source$ = (0, rxjs_1.merge)(withDisconnect$, withReconnect$);
|
62 | this.connection$ = new rxjs_1.ReplaySubject(1);
|
63 | source$.subscribe(this.connection$);
|
64 | return this.convertConnectionToPromise();
|
65 | }
|
66 | createChannel() {
|
67 | return new Promise(resolve => {
|
68 | this.channel = this.client.createChannel({
|
69 | json: false,
|
70 | setup: (channel) => this.setupChannel(channel, resolve),
|
71 | });
|
72 | });
|
73 | }
|
74 | createClient() {
|
75 | const socketOptions = this.getOptionsProp(this.options, 'socketOptions');
|
76 | return rqmPackage.connect(this.urls, {
|
77 | connectionOptions: socketOptions,
|
78 | });
|
79 | }
|
80 | mergeDisconnectEvent(instance, source$) {
|
81 | const eventToError = (eventType) => (0, rxjs_1.fromEvent)(instance, eventType).pipe((0, operators_1.map)((err) => {
|
82 | throw err;
|
83 | }));
|
84 | const disconnect$ = eventToError(constants_1.DISCONNECT_EVENT);
|
85 | const urls = this.getOptionsProp(this.options, 'urls', []);
|
86 | const connectFailed$ = eventToError(constants_1.CONNECT_FAILED_EVENT).pipe((0, operators_1.retryWhen)(e => e.pipe((0, operators_1.scan)((errorCount, error) => {
|
87 | if (urls.indexOf(error.url) >= urls.length - 1) {
|
88 | throw error;
|
89 | }
|
90 | return errorCount + 1;
|
91 | }, 0))));
|
92 |
|
93 |
|
94 | return (0, rxjs_1.merge)(source$, disconnect$, connectFailed$).pipe((0, operators_1.first)());
|
95 | }
|
96 | async convertConnectionToPromise() {
|
97 | try {
|
98 | return await (0, rxjs_1.firstValueFrom)(this.connection$);
|
99 | }
|
100 | catch (err) {
|
101 | if (err instanceof rxjs_1.EmptyError) {
|
102 | return;
|
103 | }
|
104 | throw err;
|
105 | }
|
106 | }
|
107 | async setupChannel(channel, resolve) {
|
108 | const prefetchCount = this.getOptionsProp(this.options, 'prefetchCount') ||
|
109 | constants_1.RQM_DEFAULT_PREFETCH_COUNT;
|
110 | const isGlobalPrefetchCount = this.getOptionsProp(this.options, 'isGlobalPrefetchCount') ||
|
111 | constants_1.RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
|
112 | if (!this.queueOptions.noAssert) {
|
113 | await channel.assertQueue(this.queue, this.queueOptions);
|
114 | }
|
115 | await channel.prefetch(prefetchCount, isGlobalPrefetchCount);
|
116 | await this.consumeChannel(channel);
|
117 | resolve();
|
118 | }
|
119 | async consumeChannel(channel) {
|
120 | const noAck = this.getOptionsProp(this.options, 'noAck', constants_1.RQM_DEFAULT_NOACK);
|
121 | await channel.consume(this.replyQueue, (msg) => this.responseEmitter.emit(msg.properties.correlationId, msg), {
|
122 | noAck,
|
123 | });
|
124 | }
|
125 | handleError(client) {
|
126 | client.addListener(constants_1.ERROR_EVENT, (err) => this.logger.error(err));
|
127 | }
|
128 | handleDisconnectError(client) {
|
129 | client.addListener(constants_1.DISCONNECT_EVENT, (err) => {
|
130 | this.logger.error(constants_1.DISCONNECTED_RMQ_MESSAGE);
|
131 | this.logger.error(err);
|
132 | });
|
133 | }
|
134 | async handleMessage(packet, options, callback) {
|
135 | if ((0, shared_utils_1.isFunction)(options)) {
|
136 | callback = options;
|
137 | options = undefined;
|
138 | }
|
139 | const { err, response, isDisposed } = await this.deserializer.deserialize(packet, options);
|
140 | if (isDisposed || err) {
|
141 | callback({
|
142 | err,
|
143 | response,
|
144 | isDisposed: true,
|
145 | });
|
146 | }
|
147 | callback({
|
148 | err,
|
149 | response,
|
150 | });
|
151 | }
|
152 | publish(message, callback) {
|
153 | try {
|
154 | const correlationId = (0, random_string_generator_util_1.randomStringGenerator)();
|
155 | const listener = ({ content, options, }) => this.handleMessage(this.parseMessageContent(content), options, callback);
|
156 | Object.assign(message, { id: correlationId });
|
157 | const serializedPacket = this.serializer.serialize(message);
|
158 | const options = serializedPacket.options;
|
159 | delete serializedPacket.options;
|
160 | this.responseEmitter.on(correlationId, listener);
|
161 | this.channel
|
162 | .sendToQueue(this.queue, Buffer.from(JSON.stringify(serializedPacket)), {
|
163 | replyTo: this.replyQueue,
|
164 | persistent: this.persistent,
|
165 | ...options,
|
166 | headers: this.mergeHeaders(options?.headers),
|
167 | correlationId,
|
168 | })
|
169 | .catch(err => callback({ err }));
|
170 | return () => this.responseEmitter.removeListener(correlationId, listener);
|
171 | }
|
172 | catch (err) {
|
173 | callback({ err });
|
174 | }
|
175 | }
|
176 | dispatchEvent(packet) {
|
177 | const serializedPacket = this.serializer.serialize(packet);
|
178 | const options = serializedPacket.options;
|
179 | delete serializedPacket.options;
|
180 | return new Promise((resolve, reject) => this.channel.sendToQueue(this.queue, Buffer.from(JSON.stringify(serializedPacket)), {
|
181 | persistent: this.persistent,
|
182 | ...options,
|
183 | headers: this.mergeHeaders(options?.headers),
|
184 | }, (err) => (err ? reject(err) : resolve())));
|
185 | }
|
186 | initializeSerializer(options) {
|
187 | this.serializer = options?.serializer ?? new rmq_record_serializer_1.RmqRecordSerializer();
|
188 | }
|
189 | mergeHeaders(requestHeaders) {
|
190 | if (!requestHeaders && !this.options?.headers) {
|
191 | return undefined;
|
192 | }
|
193 | return {
|
194 | ...this.options?.headers,
|
195 | ...requestHeaders,
|
196 | };
|
197 | }
|
198 | parseMessageContent(content) {
|
199 | const rawContent = content.toString();
|
200 | try {
|
201 | return JSON.parse(rawContent);
|
202 | }
|
203 | catch {
|
204 | return rawContent;
|
205 | }
|
206 | }
|
207 | }
|
208 | exports.ClientRMQ = ClientRMQ;
|