Provide topic identifier for system messages (#51)

* provide topic identifier in HandlesMessages

* rabbitmq-updates

* fold in topic identifiers

* revert keyof

Author: adenhertog

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

@@ -61,7 +61,8 @@ describe('ApplicationBootstrap', () => {
           It.isAny(),
           It.isAny(),
           TestCommandHandler,
-          TestCommand
+          TestCommand,
+          undefined
         ),
         Times.once()
       )

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

@@ -68,7 +68,8 @@ export class ApplicationBootstrap {
       prototype.$resolver,
       prototype.$symbol,
       handler,
-      prototype.$message
+      prototype.$message,
+      prototype.$topicIdentifier
     )
   }
 }

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

@@ -4,6 +4,7 @@ import { Logger } from '@node-ts/logger-core'
 import { TestEvent, TestEventHandler, TestCommandHandler, TestCommand } from '../test'
 import { Container, interfaces } from 'inversify'
 import { Message } from '@node-ts/bus-messages'
+import * as faker from 'faker'
 
 describe('HandlerRegistry', () => {
   let sut: HandlerRegistry
@@ -32,8 +33,8 @@ describe('HandlerRegistry', () => {
       expect(handlers).toHaveLength(1)
     })
 
-    it('should add it to the subscribed bus messages list', () => {
-      const subscription = sut.subscribedBusMessages.find(s => new s().$name === messageType.NAME)
+    it('should add it to the subscribed message subscription list', () => {
+      const subscription = sut.messageSubscriptions.find(s => new s.messageType!().$name === messageType.NAME)
       expect(subscription).toBeDefined()
     })
 
@@ -73,13 +74,15 @@ describe('HandlerRegistry', () => {
   })
 
   describe('when registering a handler for an external message', () => {
+    const identifier = faker.random.uuid()
+
     beforeEach(() => {
-      sut.register((m: Message) => m.$name === messageName, symbol, handler)
+      sut.register((m: Message) => m.$name === messageName, symbol, handler, undefined, identifier)
     })
 
-    it('should not include the message in the subscribed bus messages list', () => {
-      const messageSubscription = sut.subscribedBusMessages.find(m => new m().$name === messageName)
-      expect(messageSubscription).toBeUndefined()
+    it('should include the message in the subscribed messages list', () => {
+      const messageSubscription = sut.messageSubscriptions.find(m => m.topicIdentifier === identifier)
+      expect(messageSubscription).toBeDefined()
     })
   })
 

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

@@ -12,25 +12,11 @@ export interface HandlerRegistration<TMessage extends MessageType> {
   resolveHandler (handlerContextContainer: Container): Handler<TMessage>
 }
 
-interface HandlerBinding {
-  symbol: symbol
-  handler: HandlerType
-}
-
-interface RegisteredHandlers {
-  messageType: ClassConstructor<Message>
-  handlers: HandlerBinding[]
-}
-
-interface HandlerRegistrations {
-  [key: string]: RegisteredHandlers
-}
-
-type MessageName = string
-
-interface HandlerResolver {
+export interface HandlerResolver {
   handler: HandlerType
   symbol: symbol
+  topicIdentifier: string | undefined
+  messageType: ClassConstructor<Message> | undefined
   resolver (message: unknown): boolean
 }
 
@@ -42,9 +28,7 @@ interface HandlerResolver {
 export class HandlerRegistry {
 
   private container: Container
-  private unhandledMessages: MessageName[] = []
   private handlerResolvers: HandlerResolver[] = []
-  private registeredBusMessages: ClassConstructor<Message>[] = []
 
   constructor (
     @inject(LOGGER_SYMBOLS.Logger) private readonly logger: Logger
@@ -57,12 +41,15 @@ export class HandlerRegistry {
    * @param symbol A unique symbol to identify the binding of the message to the function
    * @param handler The function handler to dispatch messages to as they arrive
    * @param messageType The class type of message to handle
+   * @param topicIdentifier Identifies the topic where the message is sourced from. This topic must exist
+   * before being consumed as the library assumes it's managed externally
    */
   register<TMessage extends MessageType = MessageType> (
     resolver: (message: TMessage) => boolean,
     symbol: symbol,
     handler: HandlerType,
-    messageType?: ClassConstructor<Message>
+    messageType?: ClassConstructor<Message>,
+    topicIdentifier?: string
   ): void {
 
     const handlerName = getHandlerName(handler)
@@ -84,10 +71,8 @@ export class HandlerRegistry {
       }
     }
 
-    this.handlerResolvers.push({ resolver, symbol, handler })
-    if (!!messageType) {
-      this.registeredBusMessages.push(messageType)
-    }
+    this.handlerResolvers.push({ messageType, resolver, symbol, handler, topicIdentifier })
+
     this.logger.info(
       'Handler registered',
       { messageName: messageType ? messageType.name : undefined, handler: handlerName }
@@ -151,8 +136,8 @@ export class HandlerRegistry {
   /**
    * Retrieves a list of all messages that have handler registrations
    */
-  get subscribedBusMessages (): ClassConstructor<Message>[] {
-    return this.registeredBusMessages
+  get messageSubscriptions (): HandlerResolver[] {
+    return this.handlerResolvers
   }
 
   private bindHandlers (): void {

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

@@ -16,6 +16,16 @@ export interface HandlerPrototype<T extends MessageType> {
    */
   $symbol: symbol
 
+  /**
+   * An optional topic identifier where the source message is published to. This is used to
+   * subscribe the application queue to non-application type messages like infrastructure or
+   * third party messages.
+   *
+   * This field accepts identifiers of the transport used. eg. SQS transports will accept a
+   * source SNS arn, whereas a RabbitMQ transport would use a topic name.
+   */
+  $topicIdentifier: string | undefined
+
   /**
    * The resolver to use that determines if a given message should be dispatched to the handler
    * it's attached to

packages/bus-core/src/handler/handles-message.ts

@@ -9,20 +9,52 @@ import { ClassConstructor, isClassConstructor } from '../util'
  *
  * The dispatcher will dispatch received messages to the `handle()` function of your class.
  *
- * @param messageConstructor The type of message that the function handles
- * @param customResolver A custom resolver to use to map messages to the handler. This can be used to handle
+ * @param messageConstructor A custom resolver to use to map messages to the handler. This can be used to handle
  * messages that originate in a different system or that don't conform to the `Message` conventions.
  */
 export function HandlesMessage<
   TMessage extends MessageType,
   THandler extends Handler<TMessage>,
   HandlerConstructor extends ClassConstructor<THandler>
 > (
-  resolveWith: ClassConstructor<TMessage> | ((message: TMessage) => boolean)
+  messageConstructor: ClassConstructor<TMessage>
+  ): (handlerConstructor: HandlerConstructor) => void
+
+/**
+ * Marks that the decorated class handles a particular message. When a message
+ * matching the given type is received from the underlying transport it will be dispatched
+ * to this function.
+ *
+ * The dispatcher will dispatch received messages to the `handle()` function of your class.
+ *
+ * @param resolveWith A custom resolver to use to map messages to the handler. This can be used to handle
+ * messages that originate in a different system or that don't conform to the `Message` conventions.
+ * @param topicIdentifier Identifies the topic where the message is sourced from. This topic must exist
+ * before being consumed as the library assumes it's managed externally
+ */
+export function HandlesMessage<
+  TMessage extends MessageType,
+  THandler extends Handler<TMessage>,
+  HandlerConstructor extends ClassConstructor<THandler>
+> (
+  resolveWith: (message: TMessage) => boolean,
+  topicIdentifier: string
+  ): (handlerConstructor: HandlerConstructor) => void
+
+
+
+export function HandlesMessage<
+  TMessage extends MessageType,
+  THandler extends Handler<TMessage>,
+  HandlerConstructor extends ClassConstructor<THandler>
+> (
+  resolveWith: ClassConstructor<TMessage> | ((message: TMessage) => boolean),
+  topicIdentifier?: string
   ): (handlerConstructor: HandlerConstructor) => void {
     return (handlerConstructor: HandlerConstructor) => {
     const prototype = handlerConstructor.prototype as HandlerPrototype<TMessage>
     prototype.$symbol = Symbol.for(`node-ts/bus-core/handles-message/${handlerConstructor.name}`)
+    prototype.$topicIdentifier = topicIdentifier
 
     const isBusMessage = isClassConstructor(resolveWith)
 

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

@@ -1,3 +1,3 @@
 export * from './handler-registry'
 export * from './handles-message'
-export { Handler } from './handler'
+export { Handler, MessageType } from './handler'

packages/bus-core/src/test/test-resolver-handler.ts

@@ -13,7 +13,7 @@ export class SystemEvent {
   }
 }
 
-@HandlesMessage((e: SystemEvent) => e.type === eventType)
+@HandlesMessage((e: SystemEvent) => e.type === eventType, '')
 export class TestResolverHandler {
   constructor (
     @inject(MESSAGE_LOGGER) private readonly messageLogger: MessageLogger

packages/bus-core/src/transport/memory-queue.spec.ts

@@ -3,7 +3,7 @@ import { TestCommand, TestEvent, TestCommand2, TestSystemMessage } from '../test
 import { TransportMessage } from '../transport'
 import { Mock } from 'typemoq'
 import { Logger } from '@node-ts/logger-core'
-import { HandlerRegistry } from '../handler'
+import { HandlerRegistry, HandlerResolver } from '../handler'
 import { MessageAttributes } from '@node-ts/bus-messages'
 import * as faker from 'faker'
 
@@ -13,7 +13,7 @@ const command2 = new TestCommand2()
 
 describe('MemoryQueue', () => {
   let sut: MemoryQueue
-  const handledMessageNames = [TestCommand, TestEvent]
+  const handledMessages = [TestCommand, TestEvent]
   const messageOptions = new MessageAttributes({
     correlationId: faker.random.uuid()
    })
@@ -25,8 +25,8 @@ describe('MemoryQueue', () => {
 
     const handlerRegistry = Mock.ofType<HandlerRegistry>()
     handlerRegistry
-      .setup(h => h.subscribedBusMessages)
-      .returns(() => handledMessageNames)
+      .setup(h => h.messageSubscriptions)
+      .returns(() => handledMessages.map(h => ({ messageType: h }) as {} as HandlerResolver))
 
     await sut.initialize(handlerRegistry.object)
   })

packages/bus-core/src/transport/memory-queue.ts

@@ -46,7 +46,9 @@ export class MemoryQueue implements Transport<InMemoryMessage> {
 
   async initialize (handlerRegistry: HandlerRegistry): Promise<void> {
     this.messagesWithHandlers = {}
-    handlerRegistry.subscribedBusMessages
+    handlerRegistry.messageSubscriptions
+      .filter(subscription => !!subscription.messageType)
+      .map(subscription => subscription.messageType!)
       .map(ctor => new ctor().$name)
       .forEach(messageName => this.messagesWithHandlers[messageName] = {})
   }