feat(commands): add support for command listener

Author: juanmarin-co

example/index.ts

@@ -19,12 +19,12 @@ registry.serveQuery<string, string>('cosa.traeme', request => {
 
 app
   .start()
-  .then(connection => {
-    connection.onConnect(() => console.log('Connection stablished'))
-    connection.onDisconnect(err => console.error(err))
+  .then(context => {
+    context.connection.onConnect(() => console.log('Connection stablished'))
+    context.connection.onDisconnect(err => console.error(err))
 
     const appCreated = createEvent<string>('myInstance.appCreated')
-    connection.eventBus
+    context.eventBus
       .emit(appCreated('133', 'myApp'))
       .then(() => console.log('this will succeed'))
       .catch(err => console.error(err))

src/adapter/amqp/amqp-connection.ts

@@ -4,15 +4,13 @@ import { v4 as uuid } from 'uuid'
 import { Connection } from '../../domain/model/broker.model'
 import { AmqpReceiver } from './amqp-receiver'
 import { AmqpSender } from './amqp-sender'
-import { EventBus } from '../../domain/api/emitters/event-bus'
 
 export class AmqpConnection implements Connection {
   public id: string
 
   private channel: ChannelWrapper
   private _sender: AmqpSender
   private _receiver: AmqpReceiver
-  private _eventBus: EventBus
 
   get sender() {
     return this._sender
@@ -22,10 +20,6 @@ export class AmqpConnection implements Connection {
     return this._receiver
   }
 
-  get eventBus() {
-    return this._eventBus
-  }
-
   constructor(private connectionManager: AmqpConnectionManager) {
     this.id = uuid().replace(/-/g, '')
 
@@ -35,7 +29,6 @@ export class AmqpConnection implements Connection {
 
     this._sender = new AmqpSender(this.channel)
     this._receiver = new AmqpReceiver(this.channel)
-    this._eventBus = new EventBus(this._sender, 'domainEvents')
   }
 
   onConnect(listener: (event: { host: string }) => void) {

src/domain/api/direct-gateway.model.ts

@@ -0,0 +1,15 @@
+import { Command } from '../../..'
+import { Query } from '../../model/messages.model'
+import { BrokerSender, Message } from '../../model/broker.model'
+
+export class DirectGateway {
+  constructor(private sender: BrokerSender, private topicName: string) {}
+
+  sendCommand<T>(command: Command<T>, targetName: string): Promise<void> {
+    return this.sender.publish(this.topicName, targetName, new Message(command))
+  }
+
+  requestReply<T, R>(query: Query<T>): Promise<R> {
+    return Promise.reject(`Not implemented yet`)
+  }
+}

src/domain/api/emitters/direct-gateway.ts

@@ -0,0 +1,58 @@
+import { Connection } from '../../model/broker.model'
+import { MessageListener } from '../../model/message-listener.model'
+import { Command } from '../../model/messages.model'
+import { HandlerRegistry } from '../handler-registry'
+import { ReactiveCommonsConfiguration } from '../reactive-commons'
+
+export class CommandListener implements MessageListener {
+  constructor(
+    private config: ReactiveCommonsConfiguration,
+    private connection: Connection,
+    private registry: HandlerRegistry
+  ) {}
+
+  public setupResources(): Promise<void> {
+    const queueName = `${this.config.appName}.commands`
+    const topicName = 'directMessages'
+    const routingKey = this.config.appName
+
+    return Promise.all([
+      this.declareCommandsQueue(queueName),
+      this.declareDirectMessagesTopic(topicName),
+      this.setupBindings(queueName, topicName, routingKey)
+    ]).then()
+  }
+
+  public startListening(): Promise<void> {
+    return this.connection.receiver.consume<Command<unknown>>(
+      `${this.config.appName}.commands`,
+      message => {
+        const command = message.data
+        const handler = this.registry.getCommandHandler(command.name)
+
+        return handler(command)
+      }
+    )
+  }
+
+  private declareCommandsQueue(queueName: string): Promise<void> {
+    return this.connection.sender.declareQueue({
+      name: queueName
+    })
+  }
+
+  private declareDirectMessagesTopic(topicName: string): Promise<void> {
+    return this.connection.sender.declareTopic({
+      name: topicName,
+      type: 'direct'
+    })
+  }
+
+  private setupBindings(queueName: string, topicName: string, routingKey: string): Promise<void> {
+    return this.connection.sender.bind({
+      queue: queueName,
+      topic: topicName,
+      routingKey: routingKey
+    })
+  }
+}

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

@@ -1,9 +1,10 @@
 import { Connection } from '../../model/broker.model'
+import { MessageListener } from '../../model/message-listener.model'
+import { Event } from '../../model/messages.model'
 import { HandlerRegistry } from '../handler-registry'
 import { ReactiveCommonsConfiguration } from '../reactive-commons'
-import { Event } from '../../model/messages.model'
 
-export class EventListener {
+export class EventListener implements MessageListener {
   constructor(
     private config: ReactiveCommonsConfiguration,
     private connection: Connection,

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

@@ -1,10 +1,11 @@
 import { Connection, Headers, Message } from '../../model/broker.model'
+import { COMPLETION_ONLY_SIGNAL, CORRELATION_ID, REPLY_ID } from '../../model/headers.model'
+import { MessageListener } from '../../model/message-listener.model'
 import { Query } from '../../model/messages.model'
 import { HandlerRegistry } from '../handler-registry'
 import { ReactiveCommonsConfiguration } from '../reactive-commons'
-import { REPLY_ID, CORRELATION_ID, COMPLETION_ONLY_SIGNAL } from '../../model/headers.model'
 
-export class QueryListener {
+export class QueryListener implements MessageListener {
   constructor(
     private config: ReactiveCommonsConfiguration,
     private connection: Connection,

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

@@ -0,0 +1,15 @@
+import { Connection } from '../model/broker.model'
+import { DirectGateway } from './emitters/direct-gateway'
+import { EventBus } from './emitters/event-bus'
+
+export class ReactiveCommonsContext {
+  public id: string
+  public eventBus: EventBus
+  public directGateway: DirectGateway
+
+  constructor(public connection: Connection) {
+    this.id = connection.id
+    this.eventBus = new EventBus(connection.sender, 'domainEvents')
+    this.directGateway = new DirectGateway(connection.sender, 'directMessages')
+  }
+}

src/domain/api/reactive-commons-context.ts

@@ -1,7 +1,10 @@
+import { Broker } from '../model/broker.model'
+import { MessageListener } from '../model/message-listener.model'
 import { HandlerRegistry } from './handler-registry'
-import { Broker, Connection } from '../model/broker.model'
+import { CommandListener } from './listeners/command-listener'
 import { EventListener } from './listeners/event-listener'
 import { QueryListener } from './listeners/query-listener'
+import { ReactiveCommonsContext } from './reactive-commons-context'
 
 export interface ReactiveCommonsConfiguration {
   appName: string
@@ -14,17 +17,21 @@ export class ReactiveCommons {
     private config: ReactiveCommonsConfiguration
   ) {}
 
-  start(): Promise<Connection> {
+  start(): Promise<ReactiveCommonsContext> {
     const connection = this.broker.connect()
 
-    const eventListener = new EventListener(this.config, connection, this.registry)
-    const queryListener = new QueryListener(this.config, connection, this.registry)
+    const messageListeners: MessageListener[] = [
+      new EventListener(this.config, connection, this.registry),
+      new QueryListener(this.config, connection, this.registry),
+      new CommandListener(this.config, connection, this.registry)
+    ]
 
-    return Promise.all([
-      eventListener.setupResources(),
-      queryListener.setupResources(),
-      eventListener.startListening(),
-      queryListener.startListening()
-    ]).then(() => connection)
+    const context = new ReactiveCommonsContext(connection)
+
+    return Promise.all(
+      messageListeners.map(listener =>
+        Promise.all([listener.setupResources(), listener.startListening()])
+      )
+    ).then(() => context)
   }
 }

src/domain/api/reactive-commons.ts

@@ -5,7 +5,6 @@ export interface Connection {
   id: string
   sender: BrokerSender
   receiver: BrokerReceiver
-  eventBus: EventBus
 
   onConnect(listener: (event: { host: string }) => void): void
   onDisconnect(listener: (err: Error) => void): void