.circleci/config.yml

@@ -3,7 +3,7 @@ version: 2
 defaults: &defaults
   working_directory: ~/repo
   docker:
-    - image: circleci/node:8.14.1
+    - image: circleci/node:12.18.1
 
 jobs:
   build:

.nvmrc

@@ -1 +1 @@
-8.15.0
+12.14.1

package.json

@@ -16,9 +16,9 @@
     "docs:deploy": "./deploy-docs.sh",
     "lint": "lerna run lint",
     "test": "yarn test:unit && yarn test:integration",
-    "test:unit": "jest \"(src\\/.+\\.|/)spec\\.ts$\"",
+    "test:unit": "dotenv -e test.env -- jest \"(src\\/.+\\.|/)spec\\.ts$\"",
     "test:unit:watch": "yarn run test:unit --watch",
-    "test:integration": "jest --runInBand \"bus-(core|messages|workflow).*(src\\/.+\\.|/)integration\\.ts$\""
+    "test:integration": "dotenv -e test.env -- jest --runInBand \"bus-(core|messages|workflow).*(src\\/.+\\.|/)integration\\.ts$\""
   },
   "keywords": [
     "typescript",
@@ -44,14 +44,15 @@
   ],
   "devDependencies": {
     "@node-ts/code-standards": "^0.0.10",
-    "@types/jest": "^24.0.12",
+    "@types/jest": "^26.0.4",
+    "dotenv-cli": "^3.2.0",
     "inversify": "^5.0.1",
-    "jest": "^24.7.1",
+    "jest": "^26.1.0",
     "lerna": "^3.18.4",
     "reflect-metadata": "^0.1.13",
-    "ts-jest": "^25.1.0",
-    "tslib": "^1.9.3",
-    "typescript": "^3.3.3",
+    "ts-jest": "^26.1.1",
+    "tslib": "^2.0.0",
+    "typescript": "^3.9.6",
     "vuepress": "^1.1.0",
     "vuepress-plugin-sitemap": "^2.1.2"
   }

packages/bus-core/README.md

@@ -1,6 +1,6 @@
 # @node-ts/bus-core
 
-[![Greenkeeper badge](https://badges.greenkeeper.io/node-ts/bus.svg)](https://greenkeeper.io/)
+[![Known Vulnerabilities](https://snyk.io/test/github/node-ts/bus/badge.svg)](https://snyk.io/test/github/node-ts/bus)
 [![CircleCI](https://circleci.com/gh/node-ts/bus/tree/master.svg?style=svg)](https://circleci.com/gh/node-ts/bus/tree/master)[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://opensource.org/licenses/MIT)
 
 The core messaging framework. This package provides an in-memory queue and persistence by default, but is designed to be used with other @node-ts/bus-* packages that provide compatibility with other transports (SQS, RabbitMQ, Azure Queues) and persistence technologies (PostgreSQL, SQL Server, Oracle). 
@@ -29,7 +29,7 @@ export class ApplicationContainer extends Container {
 
 ## Register a message handler
 
-Messages are handled by defining and registring a handler class. Each time a message is received by the application, it will be dispatched to each of the registered handlers.
+Messages are handled by defining and registering a handler class. Each time a message is received by the application, it will be dispatched to each of the registered handlers.
 
 Define the handler:
 
@@ -87,4 +87,29 @@ 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)
+  const callback = message => console.log('Sending', JSON.stringify(message))
+  bus.on('send', callback)
+
+  // To remove the above hook, call bus.off():
+  bus.off('send', callback)
+}
+```
+
+## Failing a Message
+
+When an error is thrown whilst handling an error, the message is typically sent back to the queue so that it can be retried. 
+
+There are times when we know that a message will never succeed even if it were to be retried. In these situations we may not want to wait for our message to be retried before sending it to the dead letter queue, but instead bypass the retries and send it to the dead letter queue immediately.
+
+This can be done by calling `bus.fail()` from within the scope of a message handling context. This will instruct `@node-ts/bus` to forward the currently handled message to the dead letter queue and remove it from the service queue.

packages/bus-core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@node-ts/bus-core",
-  "version": "0.5.3",
+  "version": "0.6.0",
   "description": "A service bus for message-based, distributed node applications",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -13,7 +13,7 @@
     "access": "public"
   },
   "dependencies": {
-    "@node-ts/bus-messages": "^0.2.3",
+    "@node-ts/bus-messages": "^0.3.0",
     "@types/node": "^12.12.37",
     "autobind-decorator": "^2.4.0",
     "class-transformer": "^0.2.3",

packages/bus-core/src/application-bootstrap/application-bootstrap.ts

@@ -1,4 +1,3 @@
-import { Message } from '@node-ts/bus-messages'
 import { LOGGER_SYMBOLS, Logger } from '@node-ts/logger-core'
 import { Container, inject, injectable } from 'inversify'
 import { BUS_SYMBOLS } from '../bus-symbols'
@@ -86,7 +85,7 @@ export class ApplicationBootstrap {
 
   private async initializeTransport (): Promise<void> {
     if (this.transport.initialize) {
-      await this.transport.initialize(this.handlerRegistry)
+      await this.transport.initialize()
     }
   }
 }

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

@@ -2,8 +2,8 @@ import { ContainerModule, interfaces } from 'inversify'
 import { BUS_SYMBOLS, BUS_INTERNAL_SYMBOLS } from './bus-symbols'
 import { MessageAttributes } from '@node-ts/bus-messages'
 import { MemoryQueue } from './transport'
-import { ServiceBus } from './service-bus'
-import { JsonSerializer } from './serialization'
+import { ServiceBus, BusHooks } from './service-bus'
+import { JsonSerializer, MessageSerializer } from './serialization'
 import { ApplicationBootstrap } from './application-bootstrap'
 import { HandlerRegistry } from './handler'
 import { ClassConstructor } from './util'
@@ -27,6 +27,8 @@ export class BusModule extends ContainerModule {
       bindService(bind, BUS_SYMBOLS.ApplicationBootstrap, ApplicationBootstrap).inSingletonScope()
       bindService(bind, BUS_SYMBOLS.HandlerRegistry, HandlerRegistry).inSingletonScope()
       bindService(bind, BUS_SYMBOLS.JsonSerializer, JsonSerializer)
+      bindService(bind, BUS_SYMBOLS.MessageSerializer, MessageSerializer)
+      bindService(bind, BUS_INTERNAL_SYMBOLS.BusHooks, BusHooks).inSingletonScope()
 
       bind(BUS_SYMBOLS.MessageHandlingContext).toConstantValue(new MessageAttributes())
     })

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

@@ -5,9 +5,12 @@ export const BUS_SYMBOLS = {
   HandlerRegistry: Symbol.for('@node-ts/bus-core/handler-registry'),
   ApplicationBootstrap: Symbol.for('@node-ts/bus-core/application-bootstrap'),
   JsonSerializer: Symbol.for('@node-ts/bus-core/json-serializer'),
+  MessageSerializer: Symbol.for('@node-ts/bus-core/message-serializer'),
   MessageHandlingContext: Symbol.for('@node-ts/bus-core/message-handling-context')
 }
 
 export const BUS_INTERNAL_SYMBOLS = {
-  SessionScopeBinder: Symbol.for('@node-ts/bus-core/session-scope-binder')
+  SessionScopeBinder: Symbol.for('@node-ts/bus-core/session-scope-binder'),
+  BusHooks: Symbol.for('@node-ts/bus-core/bus-hooks'),
+  RawMessage: Symbol.for('@node-ts/bus-core/raw-message')
 }

packages/bus-core/src/error/fail-message-outside-handling-context.ts

@@ -0,0 +1,20 @@
+import { Message } from '@node-ts/bus-messages'
+
+export class FailMessageOutsideHandlingContext extends Error {
+  /**
+   * Calling .fail() with a message indicates that the message received from the
+   * queue can not be processed even with retries and should immediately be sent
+   * to the dead letter queue.
+   *
+   * This error occurs when .fail() has been called outside of a message handling context,
+   * or more specifically - outside the stack of a Handler() operation
+   * @param msg The message that was attempted to be failed
+   */
+  constructor (
+    readonly msg: Message
+  ) {
+    super(`Attempted to fail message outside of a message handling context`)
+    // tslint:disable-next-line:no-unsafe-any
+    Object.setPrototypeOf(this, new.target.prototype)
+  }
+}

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

@@ -0,0 +1 @@
+export * from './fail-message-outside-handling-context'

packages/bus-core/src/handler/handler-registry.spec.ts

@@ -122,4 +122,27 @@ describe('HandlerRegistry', () => {
     })
   })
 
+  describe('when getting a messageType', () => {
+
+    beforeEach(() => {
+      sut.register((m: Message) => m.$name === messageName, symbol, handler, messageType)
+    })
+
+    it('return msg constructor when found', () => {
+      const msg = {
+        $name: TestEvent.NAME
+      }
+      const result = sut.getMessageType(msg as Message)
+      expect(result).toBe(TestEvent)
+    })
+
+    it('return undefined when not found', () => {
+      const msg = {
+        $name: 'bluhbluh-very-important-command'
+      }
+      const result = sut.getMessageType(msg as Message)
+      expect(result).toBeUndefined()
+    })
+  })
+
 })

packages/bus-core/src/handler/handler-registry.ts

@@ -117,6 +117,23 @@ export class HandlerRegistry {
     }))
   }
 
+  /**
+   * Gets the type consturctor for a given message name.
+   * This is used for deserialization
+   * @param message A message instance to resolve handlers for
+   */
+  getMessageType<TMessage extends MessageType> (message: TMessage): HandlerResolver['messageType'] {
+    const resolvedHandlers = this.handlerResolvers
+      .filter(resolvers => resolvers.resolver(message))
+
+    if (resolvedHandlers.length === 0) {
+      return undefined
+    }
+
+    // Assuming one message name is only associated with one message type
+    return resolvedHandlers[0].messageType
+  }
+
   /**
    * Binds message handlers into the IoC container. All handlers should be stateless and are
    * bound in a transient scope.

packages/bus-core/src/index.ts

@@ -6,3 +6,4 @@ export * from './application-bootstrap'
 export * from './handler'
 export * from './serialization'
 export * from './util'
+export * from './error'

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

@@ -1,2 +1,3 @@
 export * from './serializer'
 export * from './json-serializer'
+export * from './message-serializer'

packages/bus-core/src/serialization/json-serializer.ts

@@ -4,18 +4,22 @@ import { ClassConstructor } from '../util'
 import { classToPlain, plainToClass, serialize, deserialize } from 'class-transformer'
 
 /**
- * A very unsafe, basic JSON serializer. This will serialize objects to strings, but will
- * deserialize strings into plain objects. These will NOT contain methods or special types,
- * so the usage of this serializer is limited.
+ * A JSON-based serializer that uses `class-transformer` to transform to and from
+ * class instances of an object rather than just their plain types. As a result,
+ * object types can use all of the serialization decorator hints provided by
+ * that library.
  */
 @injectable()
 export class JsonSerializer implements Serializer {
-  serialize<T extends object> (obj: T): string {
+  serialize<ObjectType extends object> (obj: ObjectType): string {
     return serialize(obj)
   }
 
-  deserialize<T extends object> (val: string, classConstructor: ClassConstructor<T>): T {
-    return deserialize<T> (classConstructor, val)
+  deserialize<ObjectType extends object> (
+    serialized: string,
+    classConstructor: ClassConstructor<ObjectType>
+  ): ObjectType {
+    return deserialize<ObjectType> (classConstructor, serialized)
   }
 
   toPlain<T extends object> (obj: T): object {

packages/bus-core/src/serialization/message-serializer.spec.ts

@@ -0,0 +1,74 @@
+// tslint:disable:max-classes-per-file no-inferred-empty-object-type
+import { Serializer } from './serializer'
+import { ClassConstructor } from '../util'
+import { MessageSerializer } from './message-serializer'
+import { Mock, IMock, It } from 'typemoq'
+import { HandlerRegistry } from '../handler'
+import * as faker from 'faker'
+import { Message } from '@node-ts/bus-messages'
+
+class DummyMessage {
+  $name = 'bluh'
+  $version = 1
+  value: string
+
+  constructor (s: string) {
+    this.value = s
+  }
+}
+
+class ToxicSerializer implements Serializer {
+
+  serialize<ObjectType extends object> (obj: ObjectType): string {
+    return (obj as Message).$name
+  }
+
+  deserialize<ObjectType extends object> (serialized: string, classType: ClassConstructor<ObjectType>): ObjectType {
+    return new classType(serialized)
+  }
+}
+
+describe('MessageSerializer', () => {
+
+  let sut: MessageSerializer
+  let handlerRegistry: IMock<HandlerRegistry>
+  const msgName = 'dummy-message'
+
+  beforeEach(() => {
+    handlerRegistry = Mock.ofType<HandlerRegistry>()
+    handlerRegistry.setup(h => h.getMessageType(
+      It.isObjectWith({
+        $name: msgName
+      })
+    )).returns(() => DummyMessage)
+
+    const toxicSerializer = new ToxicSerializer()
+
+    sut = new MessageSerializer(
+      toxicSerializer,
+      handlerRegistry.object
+    )
+  })
+
+  it('should use underlying serializer to serialize', () => {
+    const message: Message = {
+      $name: msgName,
+      $version: 1
+    }
+    const result = sut.serialize(message)
+    expect(result).toBe(message.$name)
+  })
+
+  it('should use underlying deserializer to deserialize', () => {
+    const msg = {
+      $name: msgName,
+      text: faker.random.words()
+    }
+    const raw = JSON.stringify(msg)
+
+    const result = sut.deserialize<DummyMessage>(raw)
+
+    handlerRegistry.verifyAll()
+    expect(result.value).toBe(raw)
+  })
+})

packages/bus-core/src/serialization/message-serializer.ts

@@ -0,0 +1,40 @@
+import { injectable, inject } from 'inversify'
+import { Serializer } from './serializer'
+import { BUS_SYMBOLS } from '../bus-symbols'
+import { HandlerRegistry } from '../handler'
+import { Message } from '@node-ts/bus-messages'
+
+/**
+ * This a wrapper around the real serializer.
+ * Unlike JsonSerializer, whose sole job is parsing data,
+ * this class will do some plumbing work to look up the Handler Registry for
+ * the message constructor.
+ *
+ * Normally, transports will use this instead of the real serializer.
+ */
+@injectable()
+export class MessageSerializer {
+
+  constructor (
+    @inject(BUS_SYMBOLS.Serializer)
+      readonly serializer: Serializer,
+    @inject(BUS_SYMBOLS.HandlerRegistry)
+      readonly handlerRegistry: HandlerRegistry
+  ) {
+  }
+
+  serialize<MessageType extends Message> (message: MessageType): string {
+    return this.serializer.serialize(message)
+  }
+
+  deserialize<MessageType extends Message> (serializedMessage: string): MessageType {
+      const naiveDerializedMessage = JSON.parse(serializedMessage) as Message
+      const messageType = this.handlerRegistry.getMessageType(naiveDerializedMessage)
+
+      return (!!messageType
+        ? this.serializer.deserialize(
+          serializedMessage,
+          messageType
+        ) : naiveDerializedMessage) as MessageType
+  }
+}