Fix bus injection into Workflow

.circleci/config.yml

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

README.md

@@ -26,11 +26,6 @@ Your code stays cleaner, as only business logic related code needs to be written
 This starter project contains the following packages preconfigured and in a runnable state:
 
 * `@node-ts/bus-core` - The core service bus that manages sending, routing, dispatch, retries etc
-* `@node-ts/bus-workflow` - Enables workflow/saga definitions to be written that coordinate higher-order business processes
-* `@node-ts/logger-core` - A generic log abstraction that allows consumers to use their preferred logger
-* `@node-ts/logger-winston` - An adapter used to log messages using winston
-* `inversify` - An IoC framework for Typescript used to manage dependency injection
-
 ## Installation
 
 This repository should be cloned or forked into your own repository:
@@ -69,9 +64,10 @@ The following scripts are available as part of the starter project:
 
 * `build` - transpiles the code to `/dist`
 * `dev` - transpiles, runs and watches for code changes
-* `dev:rebuild` - transpiles and runs the app
 * `lint` - runs linting based on [@node-ts/code-standards](https://github.com/node-ts/code-standards)
 * `lint:fix` - attempts to fix linting violations
+* `test` - run all tests
+* `test:watch` - run tests in watch mode
 
 ## Running
 

jest.config.json

@@ -0,0 +1,12 @@
+{
+  "testRegex": "(src\\/.+\\.|/)(spec|integration)\\.ts$",
+  "setupFilesAfterEnv": ["<rootDir>/test/setup.ts"],
+  "transform": {
+    "^.+\\.ts?$": [
+      "esbuild-jest",
+      {
+        "sourcemap": true
+      }
+    ]
+  }
+}

package.json

@@ -5,26 +5,34 @@
   "main": "index.js",
   "repository": "https://github.com/node-ts/bus-starter.git",
   "license": "MIT",
+  "engines": {
+    "node": "14.x"
+  },
   "scripts": {
-    "build": "tsc --project tsconfig.json --outdir dist",
-    "dev": "nodemon --watch ./src -e ts --exec 'npm run dev:rebuild' dist/index.js",
-    "dev:rebuild": "npm run build && node dist/index.js",
+    "build": "esbuild ./src/index.ts --bundle --platform=node --format=cjs --sourcemap=external --outdir=dist",
+    "dev": "nodemon --watch ./src -e ts --exec 'npm run dev:rebuild'",
+    "dev:rebuild": "npm run build && node --enable-source-maps dist/index.js",
     "lint": "tslint --project tsconfig.json",
-    "lint:fix": "npm run lint --fix"
+    "lint:fix": "npm run lint --fix",
+    "test": "jest --config=jest.config.json",
+    "test:watch": "npm run test -- --watch"
   },
   "dependencies": {
-    "@node-ts/bus-core": "^1.0.0",
+    "@node-ts/bus-core": "^1.0.3",
     "@node-ts/bus-messages": "^1.0.0",
-    "@node-ts/bus-rabbitmq": "^1.0.1",
+    "@node-ts/bus-rabbitmq": "^1.0.5",
     "reflect-metadata": "^0.1.13",
     "uuid": "^7.0.3"
   },
   "devDependencies": {
     "@node-ts/code-standards": "^0.0.10",
+    "@types/jest": "^27.0.2",
     "@types/node": "^16.10.3",
     "@types/uuid": "^7.0.5",
+    "esbuild": "^0.13.8",
+    "esbuild-jest": "^0.5.0",
+    "jest": "^27.3.1",
     "nodemon": "^2.0.13",
-    "tslint": "^6.1.3",
-    "typescript": "^3.9.10"
+    "tslint": "^6.1.3"
   }
 }

src/bus.ts

@@ -0,0 +1,59 @@
+import { startSirenTestHandler, emailMaintenanceTeamHandler } from './handlers'
+import { SirenTestWorkflow } from './workflows'
+import { RabbitMqTransportConfiguration, RabbitMqTransport } from '@node-ts/bus-rabbitmq'
+import { explainInitializationError } from './error-helpers'
+import { Bus, BusInstance } from '@node-ts/bus-core'
+
+const rabbitMqConfiguration: RabbitMqTransportConfiguration = {
+  queueName: '@node-ts/bus-starter-test',
+  connectionString: 'amqp://guest:guest@0.0.0.0',
+  maxRetries: 10
+}
+
+const rabbitMq = new RabbitMqTransport(rabbitMqConfiguration)
+
+let busInstance: BusInstance | undefined
+
+/**
+ * Initializes a new instance of bus
+ */
+export const initializeBus = async (): Promise<void> => {
+  if (!!busInstance) {
+    throw new Error('Bus has already been initialized')
+  }
+
+  try {
+    busInstance = await Bus.configure()
+      .withWorkflow(SirenTestWorkflow)
+      .withHandler(startSirenTestHandler)
+      .withHandler(emailMaintenanceTeamHandler)
+      .withTransport(rabbitMq)
+      .initialize()
+  } catch (error) {
+    explainInitializationError(error)
+    throw error
+  }
+
+  await busInstance.start()
+}
+
+/**
+ * Disposes and removes the current bus instance
+ */
+export const disposeBus = async () => {
+  if (!busInstance) {
+    throw new Error('Cannot dispose bus as it has not been initialized')
+  }
+  await busInstance.dispose()
+  busInstance = undefined
+}
+
+/**
+ * Gets the initialized bus instance
+ */
+export const bus = (): BusInstance => {
+  if (!busInstance) {
+    throw new Error('Bus has not been initialized, call initializeBus() first.')
+  }
+  return busInstance
+}

src/error-helpers.ts

@@ -0,0 +1,17 @@
+/**
+ * Provided additional helpful logging if the app fails to start up
+ */
+export const explainInitializationError = (error: Error) => {
+  const rabbitNotStarted = /ECONNREFUSED/.test(error.message)
+  if (rabbitNotStarted) {
+    console.warn(
+      'RabbitMQ not running on port 5672. '
+      + 'Try running `docker run -d -p 8080:15672 -p 5672:5672 rabbitmq:3-management`'
+    )
+  }
+
+  const rabbitNotReady = /Socket closed abruptly/.test(error.message)
+  if (rabbitNotReady) {
+    console.warn('RabbitMQ is still starting up. Wait a minute and then try again.')
+  }
+}

src/handlers/email-maintenance-team-handler.ts

@@ -1,13 +1,8 @@
-import { handlerFor, BusInstance } from '@node-ts/bus-core'
-import { EmailMaintenanceTeam, MaintenanceTeamEmailed } from '../messages'
+import { handlerFor } from '@node-ts/bus-core'
+import { EmailMaintenanceTeam } from '../messages'
+import * as emailService from '../services/email-service'
 
-export const emailMaintenanceTeamHandler = (bus: () => BusInstance) => handlerFor(
+export const emailMaintenanceTeamHandler = handlerFor(
   EmailMaintenanceTeam,
-  async ({ message, sirenId }) => {
-    console.log('Sending email to maintenance team to fix siren', { message, sirenId })
-
-    // Send the email
-    const maintenanceTeamEmailed = new MaintenanceTeamEmailed(sirenId)
-    await bus().publish(maintenanceTeamEmailed)
-  }
+  async ({ message, sirenId }) => emailService.sendEmail(message, sirenId)
 )

src/handlers/start-siren-test-handler.ts

@@ -1,11 +1,12 @@
 import { StartSirenTest, SirenTestFailed, SirenTestPassed, SirenTestStarted } from '../messages'
 import { BusInstance, handlerFor } from '@node-ts/bus-core'
 import { Uuid } from '../messages/uuid'
+import { bus } from '../bus'
 
-const MAX_SIREN_TEST_DURATION = 5000
+const MAX_SIREN_TEST_DURATION = 1000
 const TEST_FAILURE_THRESHOLD = 0.5
 
-export const startSirenTestHandler = (bus: () => BusInstance) => handlerFor(
+export const startSirenTestHandler = handlerFor(
   StartSirenTest,
   async ({ sirenId }) => {
     console.log('StartSirenTest command received, starting siren test...', { sirenId })

src/index.ts

@@ -1,38 +1,14 @@
 import 'reflect-metadata'
 
-import { Bus, BusInstance } from '@node-ts/bus-core'
 import { generateUuid } from './messages/uuid'
-
 import { StartSirenTest } from './messages'
-import { startSirenTestHandler, emailMaintenanceTeamHandler } from './handlers'
-import { SirenTestWorkflow } from './workflows'
-import { RabbitMqTransportConfiguration, RabbitMqTransport } from '@node-ts/bus-rabbitmq'
-
-const rabbitMqConfiguration: RabbitMqTransportConfiguration = {
-  queueName: '@node-ts/bus-starter-test',
-  connectionString: 'amqp://guest:guest@0.0.0.0',
-  maxRetries: 10
-}
-
-const rabbitMq = new RabbitMqTransport(rabbitMqConfiguration)
-
-let bus: BusInstance
-async function initialize (): Promise<void> {
-  bus = await Bus.configure()
-    .withWorkflow(SirenTestWorkflow)
-    .withHandler(startSirenTestHandler(() => bus)) // Late bound so it's available at runtime
-    .withHandler(emailMaintenanceTeamHandler(() =>bus))
-    .withTransport(rabbitMq)
-    .initialize()
-
-  await bus.start()
-}
+import { bus, initializeBus } from './bus'
 
 async function runDemo (): Promise<void> {
-  await bus.send(new StartSirenTest(generateUuid()))
+  await bus().send(new StartSirenTest(generateUuid()))
 }
 
-initialize()
+initializeBus()
   .then(runDemo)
   .catch(err => {
     console.log(err)

src/services/email-service.ts

@@ -0,0 +1,9 @@
+import { bus } from '../bus'
+import { MaintenanceTeamEmailed } from '../messages'
+
+export const sendEmail = async (message: string, sirenId: string) => {
+  console.log('Sending email to maintenance team', { message, sirenId })
+
+  const maintenanceTeamEmailed = new MaintenanceTeamEmailed(sirenId)
+  await bus().publish(maintenanceTeamEmailed)
+}

src/services/index.ts

@@ -0,0 +1 @@
+export * from './email-service'

src/workflows/index.ts

@@ -1,2 +1,2 @@
 export * from './siren-test-workflow'
-export * from './sirent-test-workflow-data'
+export * from './siren-test-workflow-data'

src/workflows/siren-test-workflow.integration.ts

@@ -0,0 +1,32 @@
+import { MaintenanceTeamEmailed, SirenTestFailed, SirenTestStarted } from '../messages'
+import { bus, disposeBus, initializeBus } from '../bus'
+import { generateUuid } from '../messages/uuid'
+import { Listener } from '@node-ts/bus-core'
+import { Event } from '@node-ts/bus-messages'
+
+describe('SirenTestWorkflow', () => {
+
+  beforeAll(initializeBus)
+  afterAll(disposeBus)
+
+  describe('when a siren test is started', () => {
+    const sirenId = generateUuid()
+    beforeAll(async () => bus().publish(new SirenTestStarted(sirenId)))
+
+    describe('and then the siren test fails', () => {
+      it('should send an email to the maintenance team', async () => {
+        const maintenanceEmailReceived = new Promise<void>(resolve => {
+          const callback: Listener<{ event: Event }> = ({ event }) => {
+            if (event.$name === MaintenanceTeamEmailed.NAME && (event as MaintenanceTeamEmailed).sirenId === sirenId) {
+              bus().beforePublish.off(callback)
+              resolve()
+            }
+          }
+          bus().beforePublish.on(callback)
+        })
+        await bus().publish(new SirenTestFailed(sirenId))
+        await maintenanceEmailReceived
+      })
+    })
+  })
+})

src/workflows/siren-test-workflow.ts

@@ -1,15 +1,16 @@
-import { SirenTestWorkflowData } from './sirent-test-workflow-data'
-import { SirenTestStarted, SirenTestFailed, SirenTestPassed, EmailMaintenanceTeam, MaintenanceTeamEmailed } from '../messages'
-import { BusInstance, Workflow, WorkflowMapper } from '@node-ts/bus-core'
+import { SirenTestWorkflowData } from './siren-test-workflow-data'
+import {
+  SirenTestStarted,
+  SirenTestFailed,
+  SirenTestPassed,
+  EmailMaintenanceTeam,
+  MaintenanceTeamEmailed
+} from '../messages'
+import { Workflow, WorkflowMapper } from '@node-ts/bus-core'
+import { bus } from '../bus'
 
 export class SirenTestWorkflow extends Workflow<SirenTestWorkflowData> {
 
-  constructor (
-    private readonly bus: BusInstance
-  ) {
-    super()
-  }
-
   configureWorkflow (mapper: WorkflowMapper<SirenTestWorkflowData, SirenTestWorkflow>) {
     mapper
       .withState(SirenTestWorkflowData)
@@ -39,7 +40,7 @@ export class SirenTestWorkflow extends Workflow<SirenTestWorkflowData> {
       'A siren has failed its test and requires maintenance',
       sirenId
     )
-    await this.bus.send(emailMaintenanceTeam)
+    await bus().send(emailMaintenanceTeam)
     return {}
   }
 

test/setup.ts

@@ -0,0 +1 @@
+import 'reflect-metadata'