rabbit mq event bus

Author: Fortiz2305

package-lock.json

@@ -14,7 +14,7 @@
     "dev": "NODE_ENV=dev ts-node-dev --ignore-watch node_modules --inspect=0.0.0.0:9267 ./src/apps/mooc_backend/server.ts",
     "dev:backoffice:frontend": "NODE_ENV=dev ts-node-dev --ignore-watch node_modules ./src/apps/backoffice/frontend/server.ts",
     "test": "npm run test:unit && npm run test:features",
-    "test:unit": "NODE_ENV=test jest",
+    "test:unit": "NODE_ENV=test jest --watch",
     "test:features": "NODE_ENV=test cucumber-js -p default",
     "lint": "tslint src/**/*.ts{,x}",
     "start": "NODE_ENV=production node dist/src/apps/mooc_backend/server",
@@ -25,6 +25,8 @@
     "build:clean": "rm -r dist; exit 0"
   },
   "dependencies": {
+    "@types/amqplib": "^0.5.16",
+    "@types/bluebird": "^3.5.33",
     "@types/bson": "^4.0.2",
     "@types/compression": "^1.7.0",
     "@types/convict": "^5.2.1",

package.json

@@ -3,6 +3,7 @@ import { DomainEvent } from '../../../Shared/domain/DomainEvent';
 type CreateCourseDomainEventBody = {
   readonly duration: string;
   readonly name: string;
+  readonly eventName: string;
 };
 
 export class CourseCreatedDomainEvent extends DomainEvent {
@@ -33,7 +34,8 @@ export class CourseCreatedDomainEvent extends DomainEvent {
     const { name, duration } = this;
     return {
       name,
-      duration
+      duration,
+      eventName: CourseCreatedDomainEvent.EVENT_NAME
     };
   }
 

src/Contexts/Mooc/Courses/domain/CourseCreatedDomainEvent.ts

@@ -4,4 +4,5 @@ import { DomainEventSubscriber } from './DomainEventSubscriber';
 export interface EventBus {
   publish(events: Array<DomainEvent>): Promise<void>;
   addSubscribers(subscribers: Array<DomainEventSubscriber<DomainEvent>>): void;
+  start(): Promise<void>;
 }

src/Contexts/Shared/domain/EventBus.ts

@@ -10,6 +10,8 @@ export class InMemoryAsyncEventBus implements EventBus {
     this.bus = new EventEmitterBus(subscribers);
   }
 
+  async start(): Promise<void> {}
+
   async publish(events: DomainEvent[]): Promise<void> {
     this.bus.publish(events);
   }

src/Contexts/Shared/infrastructure/EventBus/InMemoryAsyncEventBus.ts

@@ -14,6 +14,8 @@ export class InMemorySyncEventBus implements EventBus {
     this.subscriptions = new Map();
   }
 
+  async start(): Promise<void> {}
+
   async publish(events: Array<DomainEvent>): Promise<void> {
     const executions: any = [];
     events.map(event => {

src/Contexts/Shared/infrastructure/EventBus/InMemorySyncEventBus.ts

@@ -0,0 +1,63 @@
+import { Connection, Message, Exchange, Queue } from 'amqp-ts';
+import config from '../../../../../apps/mooc_backend/config/config';
+import { EventBus } from '../../../domain/EventBus';
+import { DomainEvent } from '../../../domain/DomainEvent';
+import { DomainEventSubscriber } from '../../../domain/DomainEventSubscriber';
+import { DomainEventJsonDeserializer } from '../DomainEventJsonDeserializer';
+
+export default class RabbitMqEventbus implements EventBus {
+  private connection: Connection;
+  private exchange: Exchange;
+  private queue: Queue;
+  private deserializer: DomainEventJsonDeserializer;
+  private subscribers: Map<string, Array<DomainEventSubscriber<DomainEvent>>>;
+
+  constructor(deserializer: DomainEventJsonDeserializer) {
+    const rabbitMQConfig = config.get('rabbitMQ');
+    this.connection = new Connection(`amqp://${rabbitMQConfig.user}:${rabbitMQConfig.password}@${rabbitMQConfig.host}`);
+    this.exchange = this.connection.declareExchange('domain_events_fanout', 'fanout', { durable: false });
+    this.queue = this.connection.declareQueue('coursesTest', { exclusive: true });
+    this.deserializer = deserializer;
+    this.subscribers = new Map();
+  }
+
+  async start(): Promise<void> {
+    this.queue.bind(this.exchange);
+    this.queue.activateConsumer(
+      async message => {
+        const event = this.deserializer.deserialize(message.content.toString());
+        const subscribers = this.subscribers.get(event.eventName);
+        const subscribersExecutions = subscribers!.map(subscriber => subscriber.on(event));
+        await Promise.all(subscribersExecutions);
+      },
+      { noAck: true }
+    );
+  }
+
+  async publish(events: Array<DomainEvent>): Promise<void> {
+    const executions: any = [];
+    events.map(event => {
+      const message = new Message(event.toPrimitive());
+      executions.push(this.exchange.send(message));
+    });
+
+    await Promise.all(executions);
+  }
+
+  addSubscribers(subscribers: Array<DomainEventSubscriber<DomainEvent>>): void {
+    subscribers.map(subscriber => {
+      this.addSubscriber(subscriber);
+    });
+  }
+
+  private addSubscriber(subscriber: DomainEventSubscriber<DomainEvent>): void {
+    subscriber.subscribedTo().map(event => {
+      const eventName = event.EVENT_NAME;
+      if (this.subscribers.has(eventName)) {
+        this.subscribers.get(eventName)!.push(subscriber);
+      } else {
+        this.subscribers.set(eventName, [subscriber]);
+      }
+    });
+  }
+}

src/Contexts/Shared/infrastructure/EventBus/RabbitMq/RabbitMqEventBus.ts

@@ -14,6 +14,26 @@ const convictConfig = convict({
       env: 'MONGO_URL',
       default: 'mongodb://localhost:27017/dev'
     }
+  },
+  rabbitMQ: {
+    host: {
+      doc: 'The RabbitMQ connection host',
+      format: String,
+      env: 'RABBITMQ_HOST',
+      default: 'localhost'
+    },
+    user: {
+      doc: 'The RabbitMQ connection user',
+      format: String,
+      env: 'RABBITMQ_DEFAULT_USER',
+      default: 'guest'
+    },
+    password: {
+      doc: 'The RabbitMQ connection password',
+      format: String,
+      env: 'RABBITMQ_DEFAULT_PASS',
+      default: 'guest'
+    }
   }
 });
 

src/apps/mooc_backend/config/config.ts

@@ -10,8 +10,8 @@ services:
     arguments: []
 
   Mooc.shared.EventBus:
-    class: ../../../../../Contexts/Shared/infrastructure/EventBus/InMemoryAsyncEventBus
-    arguments: []
+    class: ../../../../../Contexts/Shared/infrastructure/EventBus/RabbitMq/RabbitMqEventBus
+    arguments: ['@Mooc.shared.EventBus.DomainEventJsonDeserializer']
 
   Mooc.shared.CommandHandlersInformation:
     class: ../../../../../Contexts/Shared/infrastructure/CommandBus/CommandHandlersInformation

src/apps/mooc_backend/config/dependency-injection/Shared/application.yaml

@@ -1,14 +1,15 @@
 import container from './config/dependency-injection';
-import { InMemoryAsyncEventBus } from '../../Contexts/Shared/infrastructure/EventBus/InMemoryAsyncEventBus';
 import { Definition } from 'node-dependency-injection';
 import { DomainEventSubscriber } from '../../Contexts/Shared/domain/DomainEventSubscriber';
 import { DomainEvent } from '../../Contexts/Shared/domain/DomainEvent';
+import { EventBus } from '../../Contexts/Shared/domain/EventBus';
 
 export function registerSubscribers() {
-  const eventBus = container.get('Mooc.shared.EventBus') as InMemoryAsyncEventBus;
+  const eventBus = container.get('Mooc.shared.EventBus') as EventBus;
   const subscriberDefinitions = container.findTaggedServiceIds('domainEventSubscriber') as Map<String, Definition>;
   const subscribers: Array<DomainEventSubscriber<DomainEvent>> = [];
 
   subscriberDefinitions.forEach((value: any, key: any) => subscribers.push(container.get(key)));
   eventBus.addSubscribers(subscribers);
+  eventBus.start();
 }