amirho/nest
1
0
Fork 0
mirror of https://github.com/nestjs/nest.git synced 2026-02-21 23:11:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(microservices): add status, unwrap, on, and other features

Browse Source
This commit is contained in:
Kamil Myśliwiec
2024-11-15 10:03:05 +01:00
parent bc4667c15a
commit 3dfc7fc68e
62 changed files with 2114 additions and 852 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
packages/microservices/server/server-rmq.ts
View File

@@ -5,10 +5,7 @@ import {
} from '@nestjs/common/utils/shared.utils';
import {
CONNECTION_FAILED_MESSAGE,
CONNECT_EVENT,
CONNECT_FAILED_EVENT,
DISCONNECTED_RMQ_MESSAGE,
DISCONNECT_EVENT,
NO_MESSAGE_HANDLER,
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT,
RQM_DEFAULT_NOACK,
@@ -22,8 +19,9 @@ import {
} from '../constants';
import { RmqContext } from '../ctx-host';
import { Transport } from '../enums';
import { RmqEvents, RmqEventsMap, RmqStatus } from '../events/rmq.events';
import { RmqUrl } from '../external/rmq-url.interface';
import { CustomTransportStrategy, RmqOptions } from '../interfaces';
import { RmqOptions } from '../interfaces';
import {
IncomingRequest,
OutgoingResponse,
@@ -32,45 +30,50 @@ import {
import { RmqRecordSerializer } from '../serializers/rmq-record.serializer';
import { Server } from './server';
let rmqPackage: any = {};
// To enable type safety for RMQ. This cant be uncommented by default
// because it would require the user to install the amqplib package even if they dont use RabbitMQ
// Otherwise, TypeScript would fail to compile the code.
//
type AmqpConnectionManager =
import('amqp-connection-manager').AmqpConnectionManager;
type ChannelWrapper = import('amqp-connection-manager').ChannelWrapper;
type Message = import('amqplib').Message;
// type AmqpConnectionManager = any;
// type ChannelWrapper = any;
// type Message = any;
let rmqPackage = {} as any; // as typeof import('amqp-connection-manager');
const INFINITE_CONNECTION_ATTEMPTS = -1;
/**
* @publicApi
*/
export class ServerRMQ extends Server implements CustomTransportStrategy {
export class ServerRMQ extends Server<RmqEvents, RmqStatus> {
public readonly transportId = Transport.RMQ;
protected server: any = null;
protected channel: any = null;
protected server: AmqpConnectionManager = null;
protected channel: ChannelWrapper = null;
protected connectionAttempts = 0;
protected readonly urls: string[] | RmqUrl[];
protected readonly queue: string;
protected readonly prefetchCount: number;
protected readonly noAck: boolean;
protected readonly queueOptions: any;
protected readonly isGlobalPrefetchCount: boolean;
protected readonly noAssert: boolean;
protected pendingEventListeners: Array<{
event: keyof RmqEvents;
callback: RmqEvents[keyof RmqEvents];
}> = [];
constructor(protected readonly options: RmqOptions['options']) {
super();
this.urls = this.getOptionsProp(this.options, 'urls') || [RQM_DEFAULT_URL];
this.queue =
this.getOptionsProp(this.options, 'queue') || RQM_DEFAULT_QUEUE;
this.prefetchCount =
this.getOptionsProp(this.options, 'prefetchCount') ||
RQM_DEFAULT_PREFETCH_COUNT;
this.noAck = this.getOptionsProp(this.options, 'noAck', RQM_DEFAULT_NOACK);
this.isGlobalPrefetchCount =
this.getOptionsProp(this.options, 'isGlobalPrefetchCount') ||
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
this.queueOptions =
this.getOptionsProp(this.options, 'queueOptions') ||
RQM_DEFAULT_QUEUE_OPTIONS;
this.noAssert =
this.getOptionsProp(this.options, 'noAssert') ??
this.queueOptions.noAssert ??
RQM_DEFAULT_NO_ASSERT;
this.loadPackage('amqplib', ServerRMQ.name, () => require('amqplib'));
rmqPackage = this.loadPackage(
@@ -96,16 +99,18 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
public close(): void {
this.channel && this.channel.close();
this.server && this.server.close();
this.pendingEventListeners = [];
}
public async start(
callback?: (err?: unknown, ...optionalParams: unknown[]) => void,
) {
this.server = this.createClient();
this.server.on(CONNECT_EVENT, () => {
this.server.once(RmqEventsMap.CONNECT, () => {
if (this.channel) {
return;
}
this._status$.next(RmqStatus.CONNECTED);
this.channel = this.server.createChannel({
json: false,
setup: (channel: any) => this.setupChannel(channel, callback),
@@ -117,11 +122,18 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
'maxConnectionAttempts',
INFINITE_CONNECTION_ATTEMPTS,
);
this.server.on(DISCONNECT_EVENT, (err: any) => {
this.logger.error(DISCONNECTED_RMQ_MESSAGE);
this.logger.error(err);
});
this.server.on(CONNECT_FAILED_EVENT, (error: Record<string, unknown>) => {
this.registerConnectListener();
this.registerDisconnectListener();
this.pendingEventListeners.forEach(({ event, callback }) =>
this.server.on(event, callback),
);
this.pendingEventListeners = [];
const connectFailedEvent = 'connectFailed';
this.server.once(connectFailedEvent, (error: Record<string, unknown>) => {
this._status$.next(RmqStatus.DISCONNECTED);
this.logger.error(CONNECTION_FAILED_MESSAGE);
if (error?.err) {
this.logger.error(error.err);
@@ -149,11 +161,41 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
});
}
private registerConnectListener() {
this.server.on(RmqEventsMap.CONNECT, (err: any) => {
this._status$.next(RmqStatus.CONNECTED);
});
}
private registerDisconnectListener() {
this.server.on(RmqEventsMap.DISCONNECT, (err: any) => {
this._status$.next(RmqStatus.DISCONNECTED);
this.logger.error(DISCONNECTED_RMQ_MESSAGE);
this.logger.error(err);
});
}
public async setupChannel(channel: any, callback: Function) {
if (!this.noAssert) {
const noAssert =
this.getOptionsProp(this.options, 'noAssert') ??
this.queueOptions.noAssert ??
RQM_DEFAULT_NO_ASSERT;
if (!noAssert) {
await channel.assertQueue(this.queue, this.queueOptions);
}
await channel.prefetch(this.prefetchCount, this.isGlobalPrefetchCount);
const isGlobalPrefetchCount = this.getOptionsProp(
this.options,
'isGlobalPrefetchCount',
RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT,
);
const prefetchCount = this.getOptionsProp(
this.options,
'prefetchCount',
RQM_DEFAULT_PREFETCH_COUNT,
);
await channel.prefetch(prefetchCount, isGlobalPrefetchCount);
channel.consume(
this.queue,
(msg: Record<string, any>) => this.handleMessage(msg, channel),
@@ -192,7 +234,7 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
if (!handler) {
if (!this.noAck) {
this.logger.warn(RQM_NO_MESSAGE_HANDLER`${pattern}`);
this.channel.nack(rmqContext.getMessage(), false, false);
this.channel.nack(rmqContext.getMessage() as Message, false, false);
}
const status = 'error';
const noHandlerPacket = {
@@ -223,7 +265,7 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
): Promise<any> {
const handler = this.getHandlerByPattern(pattern);
if (!handler && !this.noAck) {
this.channel.nack(context.getMessage(), false, false);
this.channel.nack(context.getMessage() as Message, false, false);
return this.logger.warn(RQM_NO_EVENT_HANDLER`${pattern}`);
}
return super.handleEvent(pattern, packet, context);
@@ -244,6 +286,26 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
this.channel.sendToQueue(replyTo, buffer, { correlationId, ...options });
}
public unwrap<T>(): T {
if (!this.server) {
throw new Error(
'Not initialized. Please call the "listen"/"startAllMicroservices" method before accessing the server.',
);
}
return this.server as T;
}
public on<
EventKey extends keyof RmqEvents = keyof RmqEvents,
EventCallback extends RmqEvents[EventKey] = RmqEvents[EventKey],
>(event: EventKey, callback: EventCallback) {
if (this.server) {
this.server.addListener(event, callback);
} else {
this.pendingEventListeners.push({ event, callback });
}
}
protected initializeSerializer(options: RmqOptions['options']) {
this.serializer = options?.serializer ?? new RmqRecordSerializer();
}