Merge branch 'master' into 1507-dependabot-config

Author: William-Edwards-STFC

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@user-office-software/duo-message-broker",
-  "version": "1.8.0",
+  "version": "1.9.0",
   "description": "",
   "author": "SWAP",
   "license": "ISC",

packages/message-broker/package.json

@@ -1,13 +1,15 @@
-import { ConsumerCallback } from './index';
+import { ConsumerCallback, ListenOnOptions } from './index';
 
 // This class is used to store the callback function and whether it is registered or not.
 export class Consumer {
   callback: ConsumerCallback;
   registered: boolean;
+  options: ListenOnOptions;
 
-  constructor(callback: ConsumerCallback) {
+  constructor(callback: ConsumerCallback, options: ListenOnOptions = {}) {
     this.callback = callback;
     this.registered = false;
+    this.options = options;
   }
 
   register() {

packages/message-broker/src/consumer.ts

@@ -21,6 +21,7 @@ describe('RabbitMQMessageBroker', () => {
       bindQueue: jest.fn(),
       publish: jest.fn(),
       consume: jest.fn().mockResolvedValue({}),
+      prefetch: jest.fn(),
       on: jest.fn(),
     };
 
@@ -109,6 +110,25 @@ describe('RabbitMQMessageBroker', () => {
       await broker.listenOn(Queue.SCHEDULING_PROPOSAL, consumer);
       expect(mockAmqpChannel.consume).toHaveBeenCalledTimes(2);
     });
+
+    it('should call channel.prefetch with the given value before consume when prefetch option is set', async () => {
+      await broker.listenOn(Queue.SCHEDULING_PROPOSAL, consumer, {
+        prefetch: 10,
+      });
+
+      expect(mockAmqpChannel.prefetch).toHaveBeenCalledWith(10, false);
+      const prefetchOrder = (mockAmqpChannel.prefetch as jest.Mock).mock
+        .invocationCallOrder[0];
+      const consumeOrder = (mockAmqpChannel.consume as jest.Mock).mock
+        .invocationCallOrder[0];
+      expect(prefetchOrder).toBeLessThan(consumeOrder);
+    });
+
+    it('should not call channel.prefetch when prefetch option is not set', async () => {
+      await broker.listenOn(Queue.SCHEDULING_PROPOSAL, consumer);
+
+      expect(mockAmqpChannel.prefetch).not.toHaveBeenCalled();
+    });
   });
 
   describe('scheduleReconnect', () => {

packages/message-broker/src/index.spec.ts

@@ -24,6 +24,10 @@ export type ConsumerCallback = (
   properties: MessageProperties
 ) => Promise<void>;
 
+export type ListenOnOptions = {
+  prefetch?: number;
+};
+
 export interface MessageBroker {
   sendMessage(queue: Queue, type: string, message: string): Promise<void>;
   sendBroadcast(queue: Queue, type: string, message: string): Promise<void>;
@@ -36,7 +40,11 @@ export interface MessageBroker {
     queueName: string,
     exchangeName: string
   ): Promise<void>;
-  listenOn(queue: Queue, cb: ConsumerCallback): Promise<void>;
+  listenOn(
+    queue: Queue,
+    cb: ConsumerCallback,
+    options?: ListenOnOptions
+  ): Promise<void>;
   listenOnBroadcast(cb: ConsumerCallback): void;
 }
 
@@ -110,12 +118,16 @@ export class RabbitMQMessageBroker implements MessageBroker {
     }
   }
 
-  async listenOn(queue: Queue, cb: ConsumerCallback) {
+  async listenOn(
+    queue: Queue,
+    cb: ConsumerCallback,
+    options: ListenOnOptions = {}
+  ) {
     if (!this.queueConsumers.has(queue)) {
       this.queueConsumers.set(queue, []);
     }
 
-    this.queueConsumers.get(queue)?.push(new Consumer(cb));
+    this.queueConsumers.get(queue)?.push(new Consumer(cb, options));
 
     if (this.channel) {
       await this.registerConsumers();
@@ -385,6 +397,10 @@ export class RabbitMQMessageBroker implements MessageBroker {
             consumer.register();
           }
 
+          if (consumer.options.prefetch !== undefined) {
+            await this.channel.prefetch(consumer.options.prefetch, false);
+          }
+
           await this.channel
             .consume(
               queue,