Add RabbitMQ retries (#58)

* add rabbitmq retries

* minor version bump on rabbit

Author: adenhertog

packages/bus-rabbitmq/README.md

@@ -28,11 +28,20 @@ container.load(new BusRabbitMqModule())
 
 const rabbitConfiguration: RabbitMqTransportConfiguration = {
   queueName: 'accounts-application-queue',
-  connectionString: 'amqp://guest:guest@localhost'
+  connectionString: 'amqp://guest:guest@localhost',
+  maxRetries: 5
 }
 container.bind(BUS_RABBITMQ_SYMBOLS.TransportConfiguration).toConstantValue(rabbitConfiguration)
 ```
 
+## Configuration Options
+
+The RabbitMQ transport has the following configuration:
+
+*  **queueName** *(required)* The name of the service queue to create and read messages from.
+*  **connectionString** *(required)* An amqp formatted connection string that's used to connect to the RabbitMQ instance
+* **maxRetries** *(optional)* The number of attempts to retry failed messages before they're routed to the dead letter queue. *Default: 10*
+
 ## Development
 
 Local development can be done with the aid of docker to run the required infrastructure. To do so, run:

packages/bus-rabbitmq/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@node-ts/bus-rabbitmq",
   "description": "A RabbitMQ transport adapter for @node-ts/bus-core.",
-  "version": "0.4.1",
+  "version": "0.5.0",
   "license": "MIT",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

packages/bus-rabbitmq/src/rabbitmq-transport-configuration.ts

@@ -9,4 +9,10 @@ export interface RabbitMqTransportConfiguration {
    * @example amqp://guest:guest@localhost
    */
   connectionString: string
+
+  /**
+   * The maximum number of attempts to retry a failed message before routing it to the dead letter queue.
+   * @default 10
+   */
+  maxRetries?: number
 }

packages/bus-rabbitmq/src/rabbitmq-transport.integration.ts

@@ -1,5 +1,14 @@
-import { RabbitMqTransport } from './rabbitmq-transport'
-import { TestContainer, TestEvent, TestCommand, TestCommandHandler } from '../test'
+import { RabbitMqTransport, DEFAULT_MAX_RETRIES } from './rabbitmq-transport'
+import {
+  TestContainer,
+  TestEvent,
+  TestCommand,
+  TestCommandHandler,
+  TestPoisonedMessageHandler,
+  TestPoisonedMessage,
+  HANDLE_CHECKER,
+  HandleChecker
+} from '../test'
 import { BUS_RABBITMQ_INTERNAL_SYMBOLS, BUS_RABBITMQ_SYMBOLS } from './bus-rabbitmq-symbols'
 import { Connection, Channel, Message as RabbitMqMessage } from 'amqplib'
 import { TransportMessage, BUS_SYMBOLS, ApplicationBootstrap, Bus } from '@node-ts/bus-core'
@@ -8,14 +17,16 @@ import * as faker from 'faker'
 import { MessageAttributes } from '@node-ts/bus-messages'
 import { TestSystemMessage } from '../test/test-system-message'
 import { TestSystemMessageHandler } from '../test/test-system-message-handler'
+import { Mock, IMock, It, Times } from 'typemoq'
 
 export async function sleep (timeoutMs: number): Promise<void> {
   return new Promise(resolve => setTimeout(resolve, timeoutMs))
 }
 
 const configuration: RabbitMqTransportConfiguration = {
   queueName: 'node-ts/bus-rabbitmq-test',
-  connectionString: 'amqp://guest:guest@0.0.0.0'
+  connectionString: 'amqp://guest:guest@0.0.0.0',
+  maxRetries: 3
 }
 
 describe('RabbitMqTransport', () => {
@@ -25,16 +36,20 @@ describe('RabbitMqTransport', () => {
   let channel: Channel
   let container: TestContainer
   let bootstrap: ApplicationBootstrap
+  let handleChecker: IMock<HandleChecker>
 
   beforeAll(async () => {
+    handleChecker = Mock.ofType<HandleChecker>()
     container = new TestContainer()
+    container.bind(HANDLE_CHECKER).toConstantValue(handleChecker.object)
     container.bind(BUS_RABBITMQ_SYMBOLS.TransportConfiguration).toConstantValue(configuration)
     bus = container.get(BUS_SYMBOLS.Bus)
     sut = container.get(BUS_SYMBOLS.Transport)
 
     bootstrap = container.get<ApplicationBootstrap>(BUS_SYMBOLS.ApplicationBootstrap)
     bootstrap.registerHandler(TestCommandHandler)
     bootstrap.registerHandler(TestSystemMessageHandler)
+    bootstrap.registerHandler(TestPoisonedMessageHandler)
 
     const connectionFactory = container.get<() => Promise<Connection>>(BUS_RABBITMQ_INTERNAL_SYMBOLS.AmqpFactory)
     connection = await connectionFactory()
@@ -71,6 +86,35 @@ describe('RabbitMqTransport', () => {
       })
     })
 
+    describe('when retrying a poisoned message', () => {
+      const poisonedMessage = new TestPoisonedMessage(faker.random.uuid())
+      beforeAll(async () => {
+        jest.setTimeout(10000)
+        await bus.publish(poisonedMessage)
+        await new Promise<void>(resolve => {
+          channel.consume('dead-letter', msg => {
+            if (!msg) {
+              return
+            }
+
+            channel.ack(msg)
+
+            const message = JSON.parse(msg.content.toString()) as TestPoisonedMessage
+            if (message.id === poisonedMessage.id) {
+              resolve()
+            }
+          })
+        })
+      })
+
+      it(`it should fail after configuration.maxRetries attempts`, () => {
+        handleChecker.verify(
+          h => h.check(It.is<TestPoisonedMessage>(m => m.id === poisonedMessage.id), It.isAny()),
+          Times.exactly(configuration.maxRetries!)
+        )
+      })
+    })
+
     describe('when sending a command', () => {
       const command = new TestCommand()
       beforeEach(async () => {
@@ -91,23 +135,21 @@ describe('RabbitMqTransport', () => {
       const command = new TestSystemMessage()
       const channelName = command.name
 
-      beforeEach(async () => {
-        await channel.assertExchange(command.name, 'fanout')
-        await channel.bindQueue(configuration.queueName, command.name, '')
+      beforeAll(async () => {
+        jest.setTimeout(10000)
+        channel.publish(channelName, '', Buffer.from(JSON.stringify(command)))
+        await sleep(5000)
       })
 
-      afterEach(async () => {
+      afterAll(async () => {
         await channel.deleteExchange(command.name)
       })
 
       it('should handle system messages', async () => {
-        channel.publish(channelName, '', Buffer.from(JSON.stringify(command)))
-        jest.setTimeout(10000)
-        await sleep(5000)
-        const message = await sut.readNextMessage()
-
-        expect(message).toBeDefined()
-        expect(message!.domainMessage).toMatchObject(command)
+        handleChecker.verify(
+          h => h.check(It.is<TestSystemMessage>(m => m.name === command.name), It.isAny()),
+          Times.once()
+        )
       })
     })
 

packages/bus-rabbitmq/src/rabbitmq-transport.ts

@@ -7,6 +7,7 @@ import { LOGGER_SYMBOLS, Logger } from '@node-ts/logger-core'
 import { RabbitMqTransportConfiguration } from './rabbitmq-transport-configuration'
 import { MessageType } from '@node-ts/bus-core/dist/handler/handler'
 
+export const DEFAULT_MAX_RETRIES = 10
 const deadLetterExchange = '@node-ts/bus-rabbitmq/dead-letter-exchange'
 const deadLetterQueue = 'dead-letter'
 
@@ -19,6 +20,7 @@ export class RabbitMqTransport implements Transport<RabbitMqMessage> {
   private connection: Connection
   private channel: Channel
   private assertedExchanges: { [key: string]: boolean } = {}
+  private maxRetries: number
 
   constructor (
     @inject(BUS_RABBITMQ_INTERNAL_SYMBOLS.AmqpFactory)
@@ -27,6 +29,7 @@ export class RabbitMqTransport implements Transport<RabbitMqMessage> {
       private readonly configuration: RabbitMqTransportConfiguration,
     @inject(LOGGER_SYMBOLS.Logger) private readonly logger: Logger
   ) {
+    this.maxRetries = configuration.maxRetries || DEFAULT_MAX_RETRIES
   }
 
   async initialize (handlerRegistry: HandlerRegistry): Promise<void> {
@@ -90,8 +93,16 @@ export class RabbitMqTransport implements Transport<RabbitMqMessage> {
   }
 
   async returnMessage (message: TransportMessage<RabbitMqMessage>): Promise<void> {
-    this.logger.debug('Returning message', { rawMessage: message.raw })
-    this.channel.nack(message.raw)
+    const msg = JSON.parse(message.raw.content.toString())
+    const attempt = message.raw.fields.deliveryTag
+    const meta = { attempt, message: msg, rawMessage: message.raw }
+    if (attempt >= this.maxRetries) {
+      this.logger.debug('Message retries failed, sending to dead letter queue', meta)
+      this.channel.reject(message.raw, false)
+    } else {
+      this.logger.debug('Returning message', meta)
+      this.channel.nack(message.raw)
+    }
   }
 
   private async assertExchange (messageName: string): Promise<void> {
@@ -104,7 +115,10 @@ export class RabbitMqTransport implements Transport<RabbitMqMessage> {
 
   private async bindExchangesToQueue (handlerRegistry: HandlerRegistry): Promise<void> {
     await this.createDeadLetterQueue()
-    await this.channel.assertQueue(this.configuration.queueName, { durable: true, deadLetterExchange })
+    await this.channel.assertQueue(
+      this.configuration.queueName,
+      { durable: true, deadLetterExchange }
+    )
     const subscriptionPromises = handlerRegistry.messageSubscriptions
       .map(async subscription => {
 

packages/bus-rabbitmq/test/handler-checker.ts

@@ -0,0 +1,6 @@
+import { MessageAttributes } from '@node-ts/bus-messages'
+
+export const HANDLE_CHECKER = Symbol.for('node-ts/bus-rabbitmq/integration/handle-checker')
+export interface HandleChecker {
+  check<T extends object> (message: T, attributes: MessageAttributes): void
+}

packages/bus-rabbitmq/test/index.ts

@@ -2,3 +2,6 @@ export * from './test-container'
 export * from './test-event'
 export * from './test-command'
 export * from './test-command-handler'
+export * from './test-poisoned-message'
+export * from './test-poisoned-message-handler'
+export * from './handler-checker'

packages/bus-rabbitmq/test/test-poisoned-message-handler.ts

@@ -0,0 +1,19 @@
+import { HandlesMessage } from '@node-ts/bus-core'
+import { TestPoisonedMessage } from './test-poisoned-message'
+import { HANDLE_CHECKER, HandleChecker } from './handler-checker'
+import { inject } from 'inversify'
+import { MessageAttributes } from '@node-ts/bus-messages'
+
+@HandlesMessage(TestPoisonedMessage)
+export class TestPoisonedMessageHandler {
+
+  constructor (
+    @inject(HANDLE_CHECKER) private readonly handleChecker: HandleChecker
+  ) {
+  }
+
+  async handle (message: TestPoisonedMessage, messageAttributes: MessageAttributes): Promise<void> {
+    this.handleChecker.check(message, messageAttributes)
+    throw new Error('This will be routed to the DLQ after the configuration.messageTtl retries have been exhausted')
+  }
+}

packages/bus-rabbitmq/test/test-poisoned-message.ts

@@ -0,0 +1,13 @@
+import { Message } from '@node-ts/bus-messages'
+
+export class TestPoisonedMessage extends Message {
+  static NAME = '@node-ts/bus-core/test-poisoned-message'
+  $name = TestPoisonedMessage.NAME
+  $version = 1
+
+  constructor (
+    readonly id: string
+  ) {
+    super()
+  }
+}

packages/bus-rabbitmq/test/test-system-message-handler.ts

@@ -2,11 +2,7 @@ import { HandlesMessage } from '@node-ts/bus-core'
 import { MessageAttributes } from '@node-ts/bus-messages'
 import { inject } from 'inversify'
 import { TestSystemMessage } from './test-system-message'
-
-export const HANDLE_CHECKER = Symbol.for('node-ts/bus-rabbitmq/integration/handle-checker')
-export interface HandleChecker {
-  check<T extends Object> (message: T, attributes: MessageAttributes): void
-}
+import { HANDLE_CHECKER, HandleChecker } from './handler-checker'
 
 @HandlesMessage(
   (m: TestSystemMessage) => m.name === TestSystemMessage.NAME,