Add node-mongo-kafka-emitter

Author: hovoodd

src/kafka-processor.js

@@ -81,22 +81,40 @@ class KafkaProcessor {
     }
   }
 
-  on(eventName, listener) {
-    let listeners = this.listeners.get(eventName);
+  addListener(event, listener) {
+    let listeners = this.listeners.get(event);
 
     if (!listeners) {
       listeners = new Set();
-      this.listeners.set(eventName, listeners);
+      this.listeners.set(event, listeners);
     }
 
     listeners.add(listener);
+    return this;
   }
 
-  off(eventName, listener) {
-    const listeners = this.listeners.get(eventName);
-    if (!listeners) return;
+  on(event, listener) {
+    return this.addListener(event, listener);
+  }
+
+  removeListener(event, listener) {
+    const listeners = this.listeners.get(event);
+    if (listeners) listeners.delete(listener);
+    return this;
+  }
+
+  off(event, listener) {
+    return this.removeListener(event, listener);
+  }
+
+  once(eventName, listener) {
+    const wrapper = async (...args) => {
+      this.removeListener(eventName, wrapper);
+      await listener(...args);
+    };
 
-    listeners.delete(listener);
+    this.addListener(eventName, wrapper);
+    return this;
   }
 }
 

src/kafka.js

@@ -1,35 +1,41 @@
-const { Kafka } = require('kafkajs');
+const KafkaJS = require('kafkajs');
 
 const { KafkaProcessor } = require('./kafka-processor');
 
-const kafka = new Kafka({
-  clientId: 'ship',
-  brokers: ['kafka:9092'],
-});
+class Kafka {
+  constructor(config) {
+    this.kafka = new KafkaJS.Kafka(config);
 
-const processors = {
-  user: new KafkaProcessor(kafka, { groupId: 'ship' }, { topic: 'user' }),
-};
+    this.producer = this.kafka.producer();
+    this.processors = {};
+  }
 
-async function run() {
-  await Promise.all(Object.values(processors).map((p) => p.run()));
-}
+  addProcessor(topic, consumerConfig, subscribeConfig = { topic }) {
+    if (this.processors[topic]) {
+      throw new Error(`Processor for "${topic}" topic already exists`);
+    }
+
+    this.processors[topic] = new KafkaProcessor(this.kafka, consumerConfig, subscribeConfig);
+    return this;
+  }
+
+  async run() {
+    await this.producer.connect();
+    await Promise.all(Object.values(this.processors).map((p) => p.run()));
+  }
 
-async function send({ event, data, ...record }) {
-  const producer = kafka.producer();
-  await producer.connect();
-  await producer.send({
-    ...record,
-    messages: [
-      { value: JSON.stringify({ event, data }) },
-    ],
-  });
-  await producer.disconnect();
+  async send({ event, data, ...record }) {
+    await this.producer.send({
+      ...record,
+      messages: [
+        { value: JSON.stringify({ event, data }) },
+      ],
+    });
+  }
 }
 
-module.exports = {
-  kafka,
-  processors,
-  run,
-  send,
-};
+const kafka = new Kafka({ clientId: 'ship', brokers: ['kafka:9092'] });
+
+kafka.addProcessor('user', { groupId: 'ship' });
+
+module.exports = kafka;

src/migrations/migration.schema.js

@@ -8,4 +8,4 @@ const schema = Joi.object({
     .required(),
 });
 
-module.exports = (obj) => schema.validate(obj, { allowUnknown: false });
+module.exports.validate = (obj) => schema.validate(obj, { allowUnknown: false });

src/migrations/migration.service.js

@@ -1,9 +1,9 @@
 const db = require('db');
 const fs = require('fs');
 const path = require('path');
-const validateSchema = require('./migration.schema');
+const { validate } = require('./migration.schema');
 
-const service = db.createService('__migrationVersion', { validateSchema });
+const service = db.createService('__migrationVersion', { validate });
 const migrationsPath = path.join(__dirname, 'migrations');
 const _id = 'migration_version';
 

src/migrations/migrations-log/migration-log.schema.js

@@ -16,4 +16,4 @@ const schema = Joi.object({
     .required(),
 });
 
-module.exports = (obj) => schema.validate(obj, { allowUnknown: false });
+module.exports.validate = (obj) => schema.validate(obj, { allowUnknown: false });

src/migrations/migrations-log/migration-log.service.js

@@ -1,8 +1,8 @@
 const db = require('db');
 
-const validateSchema = require('./migration-log.schema.js');
+const { validate } = require('./migration-log.schema.js');
 
-const service = db.createService('__migrationLog', { validateSchema });
+const service = db.createService('__migrationLog', { validate });
 
 service.startMigrationLog = (_id, startTime, migrationVersion) => {
   return service.atomic.findOneAndUpdate({ _id }, {

src/node-mongo-kafka-emitter.js

@@ -0,0 +1,42 @@
+class NodeMongoKafkaEmitter {
+  constructor(_topic, _kafka, _logger = console) {
+    this.topic = _topic;
+    this.kafka = _kafka;
+    this.logger = _logger;
+
+    this.processor = this.kafka.processors[this.topic];
+    if (!this.processor) {
+      throw new Error(`Processor for "${this.topic}" topic doesn't exist`);
+    }
+  }
+
+  castEvent(event) {
+    return `${this.topic}:${event}`;
+  }
+
+  async emit(event, data) {
+    try {
+      await this.kafka.send({
+        topic: this.topic,
+        event: this.castEvent(event),
+        data,
+      });
+    } catch (error) {
+      this.logger.error(error);
+    }
+  }
+
+  on(event, listener) {
+    this.processor.on(this.castEvent(event), async (message) => {
+      await listener(message.data);
+    });
+  }
+
+  once(event, listener) {
+    this.processor.once(this.castEvent(event), async (message) => {
+      await listener(message.data);
+    });
+  }
+}
+
+module.exports = NodeMongoKafkaEmitter;

src/resources/account/account.spec.js

@@ -6,9 +6,9 @@ const db = require('tests/db');
 const { USER, ERRORS } = require('tests/constants');
 const testsHelper = require('tests/tests.helper');
 const UserBuilder = require('resources/user/user.builder');
-const validateSchema = require('resources/user/user.schema');
+const { validate } = require('resources/user/user.schema');
 
-const userService = db.createService(USER.COLLECTION, { validateSchema });
+const userService = db.createService(USER.COLLECTION, { validate });
 const app = server.listen();
 
 const request = supertest.agent(app);

src/resources/token/token.schema.js

@@ -16,4 +16,4 @@ const schema = Joi.object({
   isShadow: Joi.boolean(),
 });
 
-module.exports = (obj) => schema.validate(obj, { allowUnknown: false });
+module.exports.validate = (obj) => schema.validate(obj, { allowUnknown: false });

src/resources/token/token.service.js

@@ -2,9 +2,9 @@ const db = require('db');
 const securityUtil = require('security.util');
 const { DATABASE_DOCUMENTS, TOKEN_SECURITY_LENGTH, TOKEN_TYPES } = require('app.constants');
 
-const validateSchema = require('./token.schema');
+const { validate } = require('./token.schema');
 
-const service = db.createService(DATABASE_DOCUMENTS.TOKENS, { validateSchema });
+const service = db.createService(DATABASE_DOCUMENTS.TOKENS, { validate });
 
 const createToken = async (userId, type) => {
   const value = await securityUtil.generateSecureToken(TOKEN_SECURITY_LENGTH);