Add kafka and kafka-processor

docker-compose.yml

@@ -1,5 +1,26 @@
 version: '3.6'
 services:
+  zookeeper:
+    container_name: ship-zookeeper
+    image: 'bitnami/zookeeper:latest'
+    ports:
+      - '2181:2181'
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+    networks:
+      - ship
+  kafka:
+    container_name: ship-kafka
+    image: 'bitnami/kafka:latest'
+    ports:
+      - '9092:9092'
+    environment:
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+      - ALLOW_PLAINTEXT_LISTENER=yes
+    networks:
+      - ship
+    depends_on:
+      - zookeeper
   mongo:
     container_name: ship-mongo
     image: mongo:4.2

package.json

@@ -46,6 +46,7 @@
     "bcrypt": "5.0.0",
     "google-auth-library": "6.0.6",
     "joi": "17.2.1",
+    "kafkajs": "1.14.0",
     "koa": "2.13.0",
     "koa-bodyparser": "4.3.0",
     "koa-helmet": "5.2.0",

src/app.js

@@ -8,6 +8,7 @@ process.env.APP_ENV = process.env.APP_ENV || 'development';
 
 const config = require('config');
 const logger = require('logger');
+const kafka = require('kafka');
 
 process.on('unhandledRejection', (reason, p) => {
   logger.error('Possibly Unhandled Rejection at: Promise ', p, ' reason: ', reason);
@@ -19,8 +20,11 @@ require('./config/koa')(app);
 
 require('services/socketIo.service');
 
-app.listen(config.port, () => {
-  logger.warn(`Api server listening on ${config.port}, in ${process.env.NODE_ENV} mode and ${process.env.APP_ENV} environment`);
-});
+kafka.run()
+  .then(() => {
+    app.listen(config.port, () => {
+      logger.warn(`Api server listening on ${config.port}, in ${process.env.NODE_ENV} mode and ${process.env.APP_ENV} environment`);
+    });
+  });
 
 module.exports = app;

src/kafka-processor.js

@@ -0,0 +1,103 @@
+class KafkaProcessor {
+  constructor(kafka, consumerConfig, consumerSubscribeTopic) {
+    this.kafka = kafka;
+
+    this.consumerConfig = consumerConfig;
+    this.consumer = this.kafka.consumer(this.consumerConfig);
+
+    this.consumerSubscribeTopic = consumerSubscribeTopic;
+
+    this.listeners = new Map();
+
+    this.event = {
+      START_MESSAGE_PROCESSING: 'processor:start-message-processing',
+      END_MESSAGE_PROCESSING: 'processor:end-message-processing',
+    };
+
+    this.run = this.run.bind(this);
+    this.process = this.process.bind(this);
+  }
+
+  async run() {
+    await this.consumer.connect();
+    await this.consumer.subscribe(this.consumerSubscribeTopic);
+    await this.consumer.run({
+      autoCommitInterval: 5000,
+      eachMessage: this.process,
+    });
+  }
+
+  async stop() {
+    await this.consumer.stop();
+    this.listeners = new Map();
+  }
+
+  async disconnect() {
+    await this.stop();
+    await this.consumer.disconnect();
+  }
+
+  async pause() {
+    await this.consumer.pause();
+  }
+
+  async unpause() {
+    await this.consumer.unpause();
+  }
+
+  async process({ topic, partition, message }) {
+    const data = JSON.parse(message.value.toString());
+    data.kafka = { topic, partition, message };
+
+    const startListeners = this.listeners.get(this.event.START_MESSAGE_PROCESSING);
+    if (startListeners) {
+      await Promise.all(Array.from(startListeners).map(async (listener) => {
+        await listener(data);
+      }));
+    }
+
+    const start = Date.now();
+
+    const targetListeners = this.listeners.get(data.event);
+    if (targetListeners) {
+      await Promise.all(Array.from(targetListeners).map(async (listener) => {
+        await listener(data);
+      }));
+    }
+
+    const end = Date.now();
+
+    data.metadata = {
+      startProcessingOn: start,
+      endProcessingOn: end,
+      processingDuration: end - start,
+    };
+
+    const endListeners = this.listeners.get(this.event.END_MESSAGE_PROCESSING);
+    if (endListeners) {
+      await Promise.all(Array.from(endListeners).map(async (listener) => {
+        await listener(data);
+      }));
+    }
+  }
+
+  on(eventName, listener) {
+    let listeners = this.listeners.get(eventName);
+
+    if (!listeners) {
+      listeners = new Set();
+      this.listeners.set(eventName, listeners);
+    }
+
+    listeners.add(listener);
+  }
+
+  off(eventName, listener) {
+    const listeners = this.listeners.get(eventName);
+    if (!listeners) return;
+
+    listeners.delete(listener);
+  }
+}
+
+exports.KafkaProcessor = KafkaProcessor;

src/kafka.js

@@ -0,0 +1,35 @@
+const { Kafka } = require('kafkajs');
+
+const { KafkaProcessor } = require('./kafka-processor');
+
+const kafka = new Kafka({
+  clientId: 'ship',
+  brokers: ['kafka:9092'],
+});
+
+const processors = {
+  user: new KafkaProcessor(kafka, { groupId: 'ship' }, { topic: 'user' }),
+};
+
+async function run() {
+  await Promise.all(Object.values(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();
+}
+
+module.exports = {
+  kafka,
+  processors,
+  run,
+  send,
+};

src/resources/account/signup/index.js

@@ -1,5 +1,7 @@
 const Joi = require('joi');
 
+const kafka = require('kafka');
+
 const validate = require('middlewares/validate');
 const securityUtil = require('security.util');
 const userService = require('resources/user/user.service');
@@ -70,7 +72,8 @@ async function handler(ctx) {
     securityUtil.generateSecureToken(),
   ]);
 
-  const user = await userService.create({
+  const user = {
+    createdOn: new Date().toISOString(),
     firstName: data.firstName,
     lastName: data.lastName,
     email: data.email,
@@ -80,6 +83,12 @@ async function handler(ctx) {
     oauth: {
       google: false,
     },
+  };
+
+  await kafka.send({
+    topic: 'user',
+    event: 'user:signup',
+    data: user,
   });
 
   await emailService.sendSignupWelcome({

src/resources/user/user.handler.js

@@ -1,6 +1,12 @@
+const kafka = require('kafka');
+
 const userService = require('resources/user/user.service');
 const ioEmitter = require('ioEmitter');
 
+kafka.processors.user.on('user:signup', async (message) => {
+  await userService.create(message.data);
+});
+
 userService.on('updated', ({ doc }) => {
   const roomId = `user-${doc._id}`;
   ioEmitter.to(roomId).emit('user:updated', doc);