feat(rabbitmq): adds consistent rabbitmq config

use community convention of forRoot and forRootAsync to bootstrap module

fix #34

examples/kitchen-sink/src/rabbit-example/rabbit-example.module.ts

@@ -5,18 +5,20 @@ import { MessagingService } from './messaging/messaging.service';
 
 @Module({
  imports: [
- RabbitMQModule.build({
- exchanges: [
- {
- name: 'exchange1',
- type: 'topic',
- },
- {
- name: 'exchange2',
- type: 'fanout',
- },
- ],
- uri: 'amqp:https://rabbitmq:rabbitmq@localhost:5672',
+ RabbitMQModule.forRootAsync({
+ useFactory: () => ({
+ exchanges: [
+ {
+ name: 'exchange1',
+ type: 'topic',
+ },
+ {
+ name: 'exchange2',
+ type: 'fanout',
+ },
+ ],
+ uri: 'amqp:https://rabbitmq:rabbitmq@localhost:5672',
+ }),
  }),
  RabbitExampleModule,
  ],

integration/rabbitmq/e2e/configuration.e2e-spec.ts

@@ -0,0 +1,112 @@
+import { INestApplication } from '@nestjs/common';
+import { Test, TestingModule } from '@nestjs/testing';
+import { RabbitMQModule, RabbitMQConfig } from '@nestjs-plus/rabbitmq';
+import * as amqplib from 'amqplib';
+
+const uri = 'amqp:https://rabbitmq:rabbitmq@localhost:5672';
+
+class RabbitConfig {
+ createOptions(): RabbitMQConfig {
+ return {
+ uri,
+ };
+ }
+}
+
+describe('Module Configuration', () => {
+ let app: TestingModule;
+
+ afterEach(() => jest.clearAllMocks());
+
+ describe('forRoot', () => {
+ it('should configure RabbitMQ', async () => {
+ const spy = jest.spyOn(amqplib, 'connect');
+
+ app = await Test.createTestingModule({
+ imports: [
+ RabbitMQModule.forRoot({
+ uri,
+ }),
+ ],
+ }).compile();
+
+ expect(spy).toHaveBeenCalledTimes(1);
+ expect(spy).toHaveBeenCalledWith(uri);
+ });
+ });
+
+ describe('forRootAsync', () => {
+ it('should configure RabbitMQ with useFactory', async () => {
+ const spy = jest.spyOn(amqplib, 'connect');
+
+ app = await Test.createTestingModule({
+ imports: [
+ RabbitMQModule.forRootAsync({
+ useFactory: async () => {
+ return {
+ uri,
+ };
+ },
+ }),
+ ],
+ }).compile();
+
+ expect(spy).toHaveBeenCalledTimes(1);
+ expect(spy).toHaveBeenCalledWith(uri);
+ });
+
+ it('should configure RabbitMQ with useClass', async () => {
+ const spy = jest.spyOn(amqplib, 'connect');
+
+ app = await Test.createTestingModule({
+ imports: [
+ RabbitMQModule.forRootAsync({
+ useClass: RabbitConfig,
+ }),
+ ],
+ }).compile();
+
+ expect(spy).toHaveBeenCalledTimes(1);
+ expect(spy).toHaveBeenCalledWith(uri);
+ });
+
+ it('should configure RabbitMQ with useExisting explicit provide', async () => {
+ const spy = jest.spyOn(amqplib, 'connect');
+
+ const instance = new RabbitConfig();
+
+ app = await Test.createTestingModule({
+ imports: [
+ RabbitMQModule.forRootAsync({
+ useExisting: {
+ provide: RabbitConfig,
+ value: instance,
+ },
+ }),
+ ],
+ }).compile();
+
+ expect(spy).toHaveBeenCalledTimes(1);
+ expect(spy).toHaveBeenCalledWith(uri);
+ });
+
+ it('should configure RabbitMQ with useExisting implicit provide', async () => {
+ const spy = jest.spyOn(amqplib, 'connect');
+
+ const instance = new RabbitConfig();
+
+ app = await Test.createTestingModule({
+ imports: [
+ RabbitMQModule.forRootAsync({
+ useExisting: {
+ value: instance,
+ },
+ }),
+ ],
+ }).compile();
+
+ expect(spy).toHaveBeenCalledTimes(1);
+ expect(spy).toHaveBeenCalledWith(uri);
+ });
+ });
+});

integration/rabbitmq/e2e/nack-and-requeue.e2e-spec.ts

@@ -62,14 +62,16 @@ describe('Nack and Requeue', () => {
  const moduleFixture = await Test.createTestingModule({
  providers: [SubscribeService],
  imports: [
- RabbitMQModule.build({
- exchanges: [
- {
- name: exchange,
- type: 'topic',
- },
- ],
- uri: 'amqp:https://rabbitmq:rabbitmq@localhost:5672',
+ RabbitMQModule.forRootAsync({
+ useFactory: () => ({
+ exchanges: [
+ {
+ name: exchange,
+ type: 'topic',
+ },
+ ],
+ uri: 'amqp:https://rabbitmq:rabbitmq@localhost:5672',
+ }),
  }),
  ],
  }).compile();

integration/rabbitmq/e2e/subscribe.e2e-spec.ts

@@ -35,7 +35,7 @@ describe('Rabbit Subscribe', () => {
  const moduleFixture = await Test.createTestingModule({
  providers: [SubscribeService],
  imports: [
- RabbitMQModule.build({
+ RabbitMQModule.forRoot({
  exchanges: [
  {
  name: exchange,

integration/rabbitmq/src/app.module.ts

@@ -5,14 +5,16 @@ import { RpcService } from './rpc/rpc.service';
 
 @Module({
  imports: [
- RabbitMQModule.build({
- exchanges: [
- {
- name: 'exchange1',
- type: 'topic',
- },
- ],
- uri: 'amqp:https://rabbitmq:rabbitmq@localhost:5672',
+ RabbitMQModule.forRootAsync({
+ useFactory: () => ({
+ exchanges: [
+ {
+ name: 'exchange1',
+ type: 'topic',
+ },
+ ],
+ uri: 'amqp:https://rabbitmq:rabbitmq@localhost:5672',
+ }),
  }),
  ],
  controllers: [AppController],

packages/common/src/index.ts

@@ -1 +1,2 @@
 export * from './mixins';
+export * from './options';

packages/common/src/options.ts

@@ -0,0 +1,49 @@
+import { Type } from '@nestjs/common';
+import { ModuleMetadata, Provider } from '@nestjs/common/interfaces';
+import { get } from 'lodash';
+
+export interface OptionsFactory<T> {
+ createOptions(): Promise<T> | T;
+}
+
+// type OptionsFactoryImpl<T extends OptionsFactory<T>> = T;
+
+export interface AsyncOptionsFactoryProvider<T>
+ extends Pick<ModuleMetadata, 'imports'> {
+ useExisting?: {
+ value: OptionsFactory<T>;
+ provide?: string | symbol | Type<any>;
+ };
+ useClass?: Type<OptionsFactory<T>>;
+ useFactory?: (...args: any[]) => Promise<T> | T;
+ inject?: any[];
+}
+
+export function createAsyncOptionsProvider<T>(
+ provide: string | symbol | Type<any>,
+ options: AsyncOptionsFactoryProvider<T>
+): Provider {
+ if (options.useFactory) {
+ return {
+ provide,
+ useFactory: options.useFactory,
+ inject: options.inject || []
+ };
+ }
+
+ return {
+ provide,
+ useFactory: async (optionsFactory: OptionsFactory<T>) => {
+ const options = await optionsFactory.createOptions();
+ return options;
+ },
+ inject: [
+ options.useClass ||
+ get(
+ options,
+ 'useExisting.provide',
+ (options.useExisting as any).value.constructor.name
+ )
+ ]
+ };
+}

packages/rabbitmq/README.md

@@ -42,7 +42,7 @@ import { MessagingService } from './messaging/messaging.service';
 
 @Module({
  imports: [
- RabbitMQModule.build({
+ RabbitMQModule.forRoot({
  exchanges: [
  {
  name: 'exchange1',

packages/rabbitmq/src/rabbitmq.constants.ts

@@ -1 +1,2 @@
 export const RABBIT_HANDLER = Symbol('RABBIT_HANDLER');
+export const RABBIT_CONFIG = Symbol('RABBIT_CONFIG');

packages/rabbitmq/src/rabbitmq.module.ts

@@ -1,9 +1,19 @@
 import { DiscoveryModule, DiscoveryService } from '@nestjs-plus/discovery';
-import { DynamicModule, Logger, Module, OnModuleInit } from '@nestjs/common';
+import {
+ DynamicModule,
+ Logger,
+ Module,
+ OnModuleInit,
+ Provider
+} from '@nestjs/common';
+import {
+ AsyncOptionsFactoryProvider,
+ createAsyncOptionsProvider
+} from '@nestjs-plus/common';
 import { ExternalContextCreator } from '@nestjs/core/helpers/external-context-creator';
 import { groupBy } from 'lodash';
 import { AmqpConnection } from './amqp/connection';
-import { RABBIT_HANDLER } from './rabbitmq.constants';
+import { RABBIT_HANDLER, RABBIT_CONFIG } from './rabbitmq.constants';
 import { RabbitHandlerConfig, RabbitMQConfig } from './rabbitmq.interfaces';
 
 @Module({
@@ -18,7 +28,54 @@ export class RabbitMQModule implements OnModuleInit {
  private readonly externalContextCreator: ExternalContextCreator
  ) {}
 
+ public static forRootAsync(
+ asyncOptionsFactoryProvider: AsyncOptionsFactoryProvider<RabbitMQConfig>
+ ): DynamicModule {
+ return {
+ module: RabbitMQModule,
+ exports: [AmqpConnection],
+ imports: asyncOptionsFactoryProvider.imports,
+ providers: [
+ ...this.createAsyncProviders(asyncOptionsFactoryProvider),
+ {
+ provide: AmqpConnection,
+ useFactory: async (config): Promise<AmqpConnection> => {
+ const connection = new AmqpConnection(config);
+ await connection.init();
+ const logger = new Logger(RabbitMQModule.name);
+ logger.log('Successfully connected to RabbitMQ');
+ return connection;
+ },
+ inject: [RABBIT_CONFIG]
+ }
+ ]
+ };
+ }
+
+ public static forRoot(config: RabbitMQConfig): DynamicModule {
+ return {
+ module: RabbitMQModule,
+ providers: [
+ {
+ provide: AmqpConnection,
+ useFactory: async (): Promise<AmqpConnection> => {
+ const connection = new AmqpConnection(config);
+ await connection.init();
+ const logger = new Logger(RabbitMQModule.name);
+ logger.log('Successfully connected to RabbitMQ');
+ return connection;
+ }
+ }
+ ],
+ exports: [AmqpConnection]
+ };
+ }
+
  public static build(config: RabbitMQConfig): DynamicModule {
+ const logger = new Logger(RabbitMQModule.name);
+ logger.warn(
+ 'build() is deprecated. use forRoot() or forRootAsync() to configure RabbitMQ'
+ );
  return {
  module: RabbitMQModule,
  providers: [
@@ -89,4 +146,41 @@ export class RabbitMQModule implements OnModuleInit {
  );
  }
  }
+
+ private static createAsyncProviders(
+ asyncOptionsFactoryProvider: AsyncOptionsFactoryProvider<RabbitMQConfig>
+ ): Provider[] {
+ const optionsProvider = createAsyncOptionsProvider(
+ RABBIT_CONFIG,
+ asyncOptionsFactoryProvider
+ );
+
+ if (asyncOptionsFactoryProvider.useFactory) {
+ return [optionsProvider];
+ }
+
+ if (asyncOptionsFactoryProvider.useClass) {
+ return [
+ optionsProvider,
+ {
+ provide: asyncOptionsFactoryProvider.useClass,
+ useClass: asyncOptionsFactoryProvider.useClass
+ }
+ ];
+ }
+
+ if (asyncOptionsFactoryProvider.useExisting) {
+ return [
+ optionsProvider,
+ {
+ provide:
+ asyncOptionsFactoryProvider.useExisting.provide ||
+ asyncOptionsFactoryProvider.useExisting.value.constructor.name,
+ useValue: asyncOptionsFactoryProvider.useExisting.value
+ }
+ ];
+ }
+
+ return [];
+ }
 }