test: add coverage for handler registry and event bus

Author: Juan Marín

.travis.yml

@@ -9,7 +9,7 @@ node_js:
   - '11'
   - '8'
 script:
-  - npm run test:prod && npm run build
+  - npm run build
 after_success:
   - npm run travis-deploy-once "npm run report-coverage"
 branches:

example/index.ts

@@ -7,26 +7,26 @@ const app = new ReactiveCommons(broker, registry, {
 })
 
 registry.listenEvent<string>('myInstance.appCreated', event => {
-  console.log(`Evento recibido: ${event.data}`)
+  console.log(`Event received: ${event.data}`)
   return Promise.resolve()
 })
 
-registry.serverQuery<string, string>('cosa.traeme', request => {
+registry.serveQuery<string, string>('cosa.traeme', request => {
   return new Promise((resolve, reject) => {
-    setTimeout(() => resolve('holiii ' + request.queryData), 0)
+    setTimeout(() => resolve('hi ' + request.queryData), 0)
   })
 })
 
 app
   .start()
   .then(connection => {
-    connection.onConnect(() => console.log('Nos conectamos mi pana'))
+    connection.onConnect(() => console.log('Connection stablished'))
     connection.onDisconnect(err => console.error(err))
 
     const appCreated = createEvent<string>('myInstance.appCreated')
     connection.eventBus
-      .emit(appCreated('133', 'holi'))
+      .emit(appCreated('133', 'myApp'))
       .then(() => console.log('this will succeed'))
       .catch(err => console.error(err))
   })
-  .catch(err => console.error(`Esto se puteó`, err))
+  .catch(err => console.error('Error initializing app', err))

package-lock.json

@@ -9,7 +9,7 @@ import { EventEmitter } from 'events'
 export class HandlerRegistry {
   private _commandHandlers: HandlerRegistryRoutes<CommandHandler<any>> = {}
   private _queryHandlers: HandlerRegistryRoutes<QueryHandler<any, any>> = {}
-  private _eventListeners: HandlerRegistryRoutes<EventListener<any>> = {}
+  private _eventListeners: HandlerRegistryRoutes<Array<EventListener<any>>> = {}
   private eventListenersUpdates: EventEmitter
 
   get commandHandlers() {
@@ -32,24 +32,32 @@ export class HandlerRegistry {
     return new HandlerRegistry()
   }
 
-  public getCommandHandlers<M>(name: string): CommandHandler<M>[] {
-    return this.commandHandlers[name] || []
+  public getCommandHandler<M>(name: string): CommandHandler<M> {
+    return this.commandHandlers[name]
   }
 
-  public getQueryHandlers<M, R>(resource: string): QueryHandler<M, R>[] {
-    return this.queryHandlers[resource] || []
+  public getQueryHandler<M, R>(resource: string): QueryHandler<M, R> {
+    return this.queryHandlers[resource]
   }
 
   public getEventListeners<M>(name: string): EventListener<M>[] {
     return this.eventListeners[name] || []
   }
 
   public handleCommand<M>(name: string, handler: CommandHandler<M>): void {
-    this.push(this.commandHandlers, name, handler)
+    if (this.commandHandlers[name]) {
+      throw new Error(`Command ${name} already registered`)
+    }
+
+    this.commandHandlers[name] = handler
   }
 
-  public serverQuery<M, R>(resource: string, handler: QueryHandler<M, R>): void {
-    this.push(this.queryHandlers, resource, handler)
+  public serveQuery<M, R>(resource: string, handler: QueryHandler<M, R>): void {
+    if (this.queryHandlers[resource]) {
+      throw new Error(`Query ${resource} already registered`)
+    }
+
+    this.queryHandlers[resource] = handler
   }
 
   public listenEvent<M>(name: string, listener: EventListener<M>): void {
@@ -61,7 +69,7 @@ export class HandlerRegistry {
     this.eventListenersUpdates.on('eventListenerPushed', handle)
   }
 
-  private push<H>(routes: HandlerRegistryRoutes<H>, path: string, handler: H) {
+  private push<H>(routes: HandlerRegistryRoutes<Array<H>>, path: string, handler: H) {
     if (!routes[path]) {
       routes[path] = []
     }

src/domain/api/handler-registry.ts

@@ -28,11 +28,9 @@ export class QueryListener {
       `${this.config.appName}.queries`,
       message => {
         const query = message.data
-        const handlers = this.registry
-          .getQueryHandlers(query.resource)
-          .map(handler => handler(query).then(response => this.publishResponse(message, response)))
+        const handler = this.registry.getQueryHandler(query.resource)
 
-        return Promise.all(handlers).then()
+        return handler(query).then(response => this.publishResponse(message, response))
       }
     )
   }

src/domain/api/listeners/query-listener.ts

@@ -27,50 +27,4 @@ export class ReactiveCommons {
       queryListener.startListening()
     ]).then(() => connection)
   }
-
-  private listenForEvents(connection: Connection) {
-    return connection.receiver.consume(`${this.config.appName}.subscribedEvents`, message => {
-      console.log(message)
-      return Promise.resolve()
-    })
-  }
-
-  private declareEventsTopic(connectioon: Connection): Promise<void> {
-    return connectioon.sender.declareTopic({
-      name: 'domainEvents',
-      type: 'topic'
-    })
-  }
-
-  private declareDirectMessagesTopic(connection: Connection): Promise<void> {
-    return connection.sender.declareTopic({
-      name: 'directMessages',
-      type: 'direct'
-    })
-  }
-
-  private declareCommandsQueue(connection: Connection): Promise<void> {
-    return connection.sender.declareQueue({
-      name: `${this.config.appName}.commands`
-    })
-  }
-
-  private declareQueriesQueue(connection: Connection): Promise<void> {
-    return connection.sender.declareQueue({
-      name: `${this.config.appName}.queries`
-    })
-  }
-
-  private declareEventsQueue(connection: Connection): Promise<void> {
-    return connection.sender.declareQueue({
-      name: `${this.config.appName}.subscribedEvents`
-    })
-  }
-
-  private declareReplyQueue(connection: Connection): Promise<void> {
-    return connection.sender.declareQueue({
-      name: `${this.config.appName}.${connection.id}.replies`,
-      exclusive: true
-    })
-  }
 }

src/domain/api/reactive-commons.ts

@@ -5,5 +5,5 @@ export type QueryHandler<M, R> = (message: Query<M>) => Promise<R>
 export type EventListener<M> = (message: Event<M>) => Promise<void>
 
 export type HandlerRegistryRoutes<H> = {
-  [path: string]: H[]
+  [path: string]: H
 }

src/domain/model/message-handlers.model.ts

@@ -4,10 +4,11 @@ export { AmqpReceiver } from './adapter/amqp/amqp-receiver'
 export { AmqpSender } from './adapter/amqp/amqp-sender'
 export { HandlerRegistry } from './domain/api/handler-registry'
 export { createCommand, createEvent, createQuery } from './domain/api/message-factory'
+export { ReactiveCommons } from './domain/api/reactive-commons'
+export { Message } from './domain/model/broker.model'
 export { Command, Event, Query } from './domain/model/messages.model'
 export {
   BindingSpecification,
   QueueSpecitication,
   TopicSpecification
 } from './domain/model/resources.model'
-export { ReactiveCommons } from './domain/api/reactive-commons'

src/index.ts

@@ -1,10 +1,10 @@
 import { connect } from 'amqp-connection-manager'
 import { ConnectionOptions } from 'tls'
-import { AmqpBroker } from '../../src'
-import { AmqpConnection } from '../../src/adapter/amqp/amqp-connection'
+import { AmqpBroker } from '../../../src'
+import { AmqpConnection } from '../../../src/adapter/amqp/amqp-connection'
 
 jest.mock('amqp-connection-manager')
-jest.mock('../../src/adapter/amqp/amqp-connection')
+jest.mock('../../../src/adapter/amqp/amqp-connection')
 
 const url = 'amqp://myrabbitserver'
 const options: ConnectionOptions = { timeout: 10 }

test/adapter/amqp-broker.test.ts renamed to test/adapter/amqp/amqp-broker.test.ts

@@ -1,5 +1,11 @@
 import { AmqpConnectionManager } from 'amqp-connection-manager'
-import { AmqpConnection, AmqpSender, AmqpReceiver } from '../../src'
+import { AmqpConnection, AmqpReceiver, AmqpSender } from '../../../src'
+import { EventBus } from '../../../src/domain/api/emitters/event-bus'
+
+jest.mock('../../../src/domain/api/emitters/event-bus')
+jest.mock('../../../src/adapter/amqp/amqp-sender')
+jest.mock('../../../src/adapter/amqp/amqp-receiver')
+
 
 describe(`AMQP connection`, () => {
   let createChannelMock
@@ -29,6 +35,12 @@ describe(`AMQP connection`, () => {
     expect(receiver).toBeInstanceOf(AmqpReceiver)
   })
 
+  it(`creates EventBus instance`, () => {
+    const receiver = connection.eventBus
+
+    expect(receiver).toBeInstanceOf(EventBus)
+  })
+
   it(`calls onConnect callback when connectionManager emits it`, () => {
     const url = 'amqp://myrabbitserver'