Add tests for handler errors (#42)

* add tests for handler errors
* provide selective silencing on integration test logs

packages/bus-core/src/service-bus/service-bus.integration.ts

@@ -3,39 +3,61 @@ import { MemoryQueue } from '../transport'
 import { BusState } from './bus'
 import { TestEvent } from '../test/test-event'
 import { sleep } from '../util'
-import { Container } from 'inversify'
+import { Container, inject } from 'inversify'
 import { TestContainer } from '../test/test-container'
 import { BUS_SYMBOLS } from '../bus-symbols'
 import { Logger } from '@node-ts/logger-core'
-import { Mock } from 'typemoq'
-import { HandlerRegistry } from '../handler'
+import { Mock, IMock, Times } from 'typemoq'
+import { HandlerRegistry, HandlesMessage } from '../handler'
+import { ApplicationBootstrap } from '../application-bootstrap'
 
 const event = new TestEvent()
+type Callback = () => void
+const CALLBACK = Symbol.for('Callback')
+
+@HandlesMessage(TestEvent)
+class TestEventHandler {
+  constructor (
+    @inject(CALLBACK) private readonly callback: Callback
+  ) {
+  }
+
+  async handle (_: TestEvent): Promise<void> {
+    this.callback()
+  }
+}
 
 describe('ServiceBus', () => {
   let container: Container
 
   let sut: ServiceBus
+  let bootstrapper: ApplicationBootstrap
   let queue: MemoryQueue
 
+  let callback: IMock<Callback>
+
   beforeAll(async () => {
     container = new TestContainer().silenceLogs()
     queue = new MemoryQueue(Mock.ofType<Logger>().object)
+
     const transport = container.get<MemoryQueue>(BUS_SYMBOLS.Transport)
     const registry = container.get<HandlerRegistry>(BUS_SYMBOLS.HandlerRegistry)
+
+    bootstrapper = container.get<ApplicationBootstrap>(BUS_SYMBOLS.ApplicationBootstrap)
+    bootstrapper.registerHandler(TestEventHandler)
+
+    callback = Mock.ofType<Callback>()
+    container.bind(CALLBACK).toConstantValue(callback.object)
     await transport.initialize(registry)
+    await bootstrapper.initialize(container)
     sut = container.get(BUS_SYMBOLS.Bus)
   })
 
-  describe('when starting the service bus', () => {
-    beforeAll(async () => {
-      await sut.start()
-    })
-
-    afterAll(async () => {
-      await sut.stop()
-    })
+  afterAll(async () => {
+    await bootstrapper.dispose()
+  })
 
+  describe('when starting the service bus', () => {
     it('should complete into a started state', () => {
       expect(sut.state).toEqual(BusState.Started)
     })
@@ -45,16 +67,42 @@ describe('ServiceBus', () => {
         await expect(sut.start()).rejects.toThrowError()
       })
     })
+  })
 
-    describe('and a message is received on the queue', () => {
-      beforeAll(async () => {
-        await sut.publish(event)
-        await sleep(1)
-      })
 
-      it('should delete the message from the queue', () => {
-        expect(queue.depth).toEqual(0)
-      })
+  describe('when a message is successfully handled from the queue', () => {
+    it('should delete the message from the queue', async () => {
+      callback.reset()
+      callback
+        .setup(c => c())
+        .callback(() => undefined)
+        .verifiable(Times.once())
+      await sut.publish(event)
+      await sleep(10)
+
+      expect(queue.depth).toEqual(0)
+      callback.verifyAll()
+    })
+  })
+
+  describe('and a handled message throw an Error', () => {
+
+    it('should return the message for retry', async () => {
+      callback.reset()
+      let callCount = 0
+      callback
+        .setup(c => c())
+        .callback(() => {
+          if (callCount++ === 0) {
+            throw new Error()
+          }
+        })
+        .verifiable(Times.exactly(2))
+
+      await sut.publish(event)
+      await sleep(2000)
+
+      callback.verifyAll()
     })
   })
 })

packages/bus-core/src/test/test-container.ts

@@ -1,7 +1,10 @@
+// tslint:disable:no-unsafe-any - Any used for mock assertions
+
 import { Container } from 'inversify'
-import { LoggerModule, LOGGER_SYMBOLS, Logger } from '@node-ts/logger-core'
+import { LoggerModule, LOGGER_SYMBOLS, LoggerFactory, Logger } from '@node-ts/logger-core'
 import { BusModule } from '../bus-module'
-import { Mock } from 'typemoq'
+import { Mock, It } from 'typemoq'
+import { assertUnreachable } from '../util'
 
 export class TestContainer extends Container {
 
@@ -11,8 +14,36 @@ export class TestContainer extends Container {
     this.load(new BusModule())
   }
 
-  silenceLogs (): this {
-    this.rebind(LOGGER_SYMBOLS.Logger).toConstantValue(Mock.ofType<Logger>().object)
+  /**
+   * Swallows logs during integration test runs. Useful in not cluttering up
+   * the test output.
+   *
+   * @param errorLevel The error to dispay logs for. @default 'warn'
+   */
+  silenceLogs (errorLevel: keyof Logger = 'warn'): this {
+    const factory = this.get<LoggerFactory>(LOGGER_SYMBOLS.LoggerFactory)
+    const logger = factory.build('test-logger', this)
+
+    const mockLogger = Mock.ofInstance(logger)
+    logsToOutput(errorLevel).forEach(level => {
+      mockLogger
+        .setup(l => l[level](It.isAnyString(), It.isAny()))
+        .callBase()
+    })
+
+    this.rebind(LOGGER_SYMBOLS.Logger).toConstantValue(mockLogger.object)
     return this
   }
 }
+
+function logsToOutput (errorLevel: keyof Logger): (keyof Logger)[] {
+  const levels: (keyof Logger)[] = [
+    'debug',
+    'trace',
+    'info',
+    'warn',
+    'error',
+    'fatal'
+  ]
+  return levels.slice(levels.indexOf(errorLevel), levels.length)
+}

packages/bus-core/src/util/assert-unreachable.ts

@@ -0,0 +1,6 @@
+/**
+ * This raises transpile time errors any time the tokenizer hits this function in the code
+ */
+export function assertUnreachable (unexpectedValue: never): never {
+  throw new Error(`Unexepected code path - ${unexpectedValue}`)
+}

packages/bus-core/src/util/index.ts

@@ -1,2 +1,3 @@
 export * from './sleep'
 export * from './class-constructor'
+export * from './assert-unreachable'