feat(@rhtml/amqp): added additional channel configuration

packages/amqp/README.md

@@ -41,6 +41,14 @@ import { AmqpModule } from '@rhtml/amqp';
       username: 'amqp user', // AMQP username (default: 'guest')
       password: 'amqp password', // AMQP password (default: 'guest')
       vhost: '', // Virtual host to use (default: empty string)
+      /**
+       * Optional
+       * Set the prefetch count for this channel.
+       * The count given is the maximum number of messages sent over the channel that can be awaiting acknowledgement;
+       * once there are count messages outstanding,
+       * the server will not send more messages on this channel until one or more have been acknowledged.
+       */
+      prefetchCount: 1
     }),
   ],
 })

packages/amqp/src/amqp.constants.ts

@@ -1,5 +1,5 @@
 import { InjectionToken } from '@rhtml/di';
-import { Channel, Connection } from 'amqplib';
+import { Channel, Connection, Options } from 'amqplib';
 
 /**
  * Injection for AmqpConnection
@@ -12,3 +12,13 @@ export type AmqpConnection = Connection;
  */
 export const AmqpChannel = new InjectionToken<Channel>();
 export type AmqpChannel = Channel;
+
+export interface ModuleConfig extends Options.Connect {
+  /**
+   * Set the prefetch count for this channel.
+   * The count given is the maximum number of messages sent over the channel that can be awaiting acknowledgement;
+   * once there are count messages outstanding,
+   * the server will not send more messages on this channel until one or more have been acknowledged.
+   */
+  prefetchCount?: number;
+}

packages/amqp/src/amqp.module.ts

@@ -1,14 +1,14 @@
 import { Module, ModuleWithProviders } from '@rhtml/di';
-import amqpClient, { Connection, Options } from 'amqplib';
+import amqpClient, { Connection } from 'amqplib';
 
-import { AmqpChannel, AmqpConnection } from './amqp.constants';
+import { AmqpChannel, AmqpConnection, ModuleConfig } from './amqp.constants';
 import { AmqpService } from './amqp.service';
 
 @Module({
   providers: [AmqpService],
 })
 export class AmqpModule {
-  public static forRoot(config: Options.Connect): ModuleWithProviders {
+  public static forRoot(config: ModuleConfig): ModuleWithProviders {
     return {
       module: AmqpModule,
       providers: [
@@ -19,7 +19,13 @@ export class AmqpModule {
         {
           provide: AmqpChannel,
           deps: [AmqpConnection],
-          useFactory: (connection: Connection) => connection.createChannel(),
+          useFactory: async (connection: Connection) => {
+            const channel = await connection.createChannel();
+            if (config.prefetchCount) {
+              await channel.prefetch(config.prefetchCount);
+            }
+            return channel;
+          },
         },
       ],
     };