Squashed commit of the following:

commit ae945ff1cd1b8c3f36130fec2ceb1e26f221c0c3
Author: Andrew den Hertog <andrew.denhertog@gmail.com>
Date:   Tue Jun 2 15:16:46 2020 +1000

    add hooks

Author: adenhertog

packages/bus-core/README.md

@@ -87,4 +87,17 @@ export class Application {
     await this.bus.stop()
   }
 }
+```
+
+## Hooks
+
+Hooks are callback functions that are invoked each time an action occurs. These are commonly used to add in testing, logging or health probes centrally to the application.
+
+Hooks can be added by calling `.on()` on the bus. For example, to trigger a callback each time a message is attempted to be sent, use:
+
+```typescript
+addHook (): void {
+  const bus = this.container.get<Bus>(BUS_SYMBOLS.Bus)
+  bus.on('send', message => console.log('Sending', JSON.stringify(message)))
+}
 ```

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

@@ -1,4 +1,4 @@
-import { Event, Command, MessageAttributes } from '@node-ts/bus-messages'
+import { Event, Command, MessageAttributes, Message } from '@node-ts/bus-messages'
 
 export enum BusState {
   Stopped = 'stopped',
@@ -7,9 +7,17 @@ export enum BusState {
   Stopping = 'stopping'
 }
 
+export type HookAction = 'send' | 'publish'
+export type HookCallback = (message: Message, messageAttributes?: MessageAttributes) => Promise<void>
+
 export interface Bus {
   publish<EventType extends Event> (event: EventType, messageOptions?: MessageAttributes): Promise<void>
   send<CommandType extends Command> (command: CommandType, messageOptions?: MessageAttributes): Promise<void>
   start (): Promise<void>
   stop (): Promise<void>
+
+  /**
+   * Registers a @param callback function that is invoked for every instance of @param action occuring
+   */
+  on (action: HookAction, callback: HookCallback): void
 }

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

@@ -2,14 +2,17 @@ import { ServiceBus } from './service-bus'
 import { MemoryQueue } from '../transport'
 import { BusState } from './bus'
 import { TestEvent } from '../test/test-event'
+import { TestEvent2 } from '../test/test-event-2'
+import { TestCommand } from '../test/test-command'
 import { sleep } from '../util'
 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, IMock, Times } from 'typemoq'
-import { HandlerRegistry, HandlesMessage } from '../handler'
+import { HandlesMessage } from '../handler'
 import { ApplicationBootstrap } from '../application-bootstrap'
+import { MessageAttributes } from '@node-ts/bus-messages'
 
 const event = new TestEvent()
 type Callback = () => void
@@ -40,9 +43,6 @@ describe('ServiceBus', () => {
     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)
 
@@ -56,6 +56,26 @@ describe('ServiceBus', () => {
     await bootstrapper.dispose()
   })
 
+  describe('when registering a send hook', () => {
+    it('should trigger the hook when send() is called', async () => {
+      const sendCallback = jest.fn()
+      sut.on('send', sendCallback)
+      const command = new TestCommand()
+      await sut.send(command)
+      expect(sendCallback).toHaveBeenCalledWith(command, expect.any(MessageAttributes))
+    })
+  })
+
+  describe('when registering a publish hook', () => {
+    it('should trigger the hook when publish() is called', async () => {
+      const publishCallback = jest.fn()
+      sut.on('publish', publishCallback)
+      const evt = new TestEvent2()
+      await sut.publish(evt)
+      expect(publishCallback).toHaveBeenCalledWith(evt, expect.any(MessageAttributes))
+    })
+  })
+
   describe('when starting the service bus', () => {
     it('should complete into a started state', () => {
       expect(sut.state).toEqual(BusState.Started)
@@ -68,7 +88,6 @@ describe('ServiceBus', () => {
     })
   })
 
-
   describe('when a message is successfully handled from the queue', () => {
     it('should delete the message from the queue', async () => {
       callback.reset()

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

@@ -1,11 +1,11 @@
 import { injectable, inject } from 'inversify'
 import autobind from 'autobind-decorator'
-import { Bus, BusState } from './bus'
+import { Bus, BusState, HookAction, HookCallback } from './bus'
 import { BUS_SYMBOLS, BUS_INTERNAL_SYMBOLS } from '../bus-symbols'
 import { Transport } from '../transport'
 import { Event, Command, Message, MessageAttributes } from '@node-ts/bus-messages'
 import { Logger, LOGGER_SYMBOLS } from '@node-ts/logger-core'
-import { sleep } from '../util'
+import { sleep, assertUnreachable } from '../util'
 import { HandlerRegistry, HandlerRegistration } from '../handler'
 import * as serializeError from 'serialize-error'
 import { SessionScopeBinder } from '../bus-module'
@@ -20,6 +20,11 @@ export class ServiceBus implements Bus {
   private internalState: BusState = BusState.Stopped
   private runningWorkerCount = 0
 
+  private messageHooks: { [key: string]: HookCallback[] } = {
+    send: [],
+    publish: []
+  }
+
   constructor (
     @inject(BUS_SYMBOLS.Transport) private readonly transport: Transport<{}>,
     @inject(LOGGER_SYMBOLS.Logger) private readonly logger: Logger,
@@ -34,6 +39,8 @@ export class ServiceBus implements Bus {
   ): Promise<void> {
     this.logger.debug('publish', { event })
     const transportOptions = this.prepareTransportOptions(messageOptions)
+
+    this.messageHooks.publish.map(callback => callback(event, messageOptions))
     return this.transport.publish(event, transportOptions)
   }
 
@@ -43,6 +50,8 @@ export class ServiceBus implements Bus {
   ): Promise<void> {
     this.logger.debug('send', { command })
     const transportOptions = this.prepareTransportOptions(messageOptions)
+
+    this.messageHooks.send.map(callback => callback(command, messageOptions))
     return this.transport.send(command, transportOptions)
   }
 
@@ -72,6 +81,10 @@ export class ServiceBus implements Bus {
     return this.internalState
   }
 
+  on (action: HookAction, callback: HookCallback): void {
+    this.messageHooks[action].push(callback)
+  }
+
   private async applicationLoop (): Promise<void> {
     this.runningWorkerCount++
     while (this.internalState === BusState.Started) {

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

@@ -5,3 +5,4 @@ export * from './test-container'
 export * from './test-command-handler'
 export * from './test-command-2'
 export * from './test-system-message'
+export * from './test-event-2'

packages/bus-core/src/test/test-event-2.ts

@@ -0,0 +1,7 @@
+import { Event } from '@node-ts/bus-messages'
+
+export class TestEvent2 extends Event {
+  static NAME = '@node-ts/bus-core/test-event-2'
+  $name = TestEvent2.NAME
+  $version = 1
+}