git init

Author: piyal-ict-mbstu

.env

@@ -0,0 +1,5 @@
+
+MYSQL_HOST="localhost"
+MYSQL_PORT="3306"
+MYSQL_USER="root"
+MYSQL_PASSWORD="rootPwd@2022"

.env.example

@@ -0,0 +1,4 @@
+MYSQL_HOST="0.0.0.0"
+MYSQL_PORT="3306"
+MYSQL_USER="root"
+MYSQL_PASSWORD="root"

.idea/.gitignore

@@ -0,0 +1,15 @@
+const { Kafka } = require('kafkajs')
+
+const kafka = new Kafka({
+  clientId: 'my-app',
+  brokers: ['localhost:9092']
+})
+const producer = kafka.producer()
+
+const sendMessage = async (message) => {
+    await producer.connect()
+    await producer.send(message)
+    await producer.disconnect()
+  }
+
+module.exports=sendMessage

.idea/Mysql-DB-Sinker.iml

@@ -0,0 +1,44 @@
+const mysql = require('mysql');
+const MySQLEvents = require('@rodrigogs/mysql-events');
+const sendMessage = require('../Producer/Producer');
+
+
+const program = async () => {
+  const connection = mysql.createConnection({
+    user: 'root',
+    password: 'root'
+  });
+  
+  const instance = new MySQLEvents(connection, {
+    startAtEnd: true,
+    excludedSchemas: {
+      mysql: true,
+    },
+  });
+
+  await instance.start();
+
+  instance.addTrigger({
+    name: 'OPERATIONS',
+    expression: '*',
+    statement: MySQLEvents.STATEMENTS.ALL,
+    onEvent: (event) => { // You will receive the events here
+       const message=
+        {
+          topic: event.table,
+          messages: [
+            { key: event.table, value: JSON.stringify(event) }
+          ],
+        }
+        console.log(message);
+      sendMessage(message);
+    },
+  });
+  
+  instance.on(MySQLEvents.EVENTS.CONNECTION_ERROR, console.error);
+  instance.on(MySQLEvents.EVENTS.ZONGJI_ERROR, console.error);
+};
+
+program()
+  .then(() => console.log('Waiting for database events...'))
+  .catch(console.error);

.idea/modules.xml

@@ -0,0 +1,40 @@
+//mongodb://localhost:27017
+const { MongoClient } = require("mongodb");
+
+// Replace the uri string with your MongoDB deployment's connection string.
+const uri ="mongodb://localhost:27017"
+
+const client = new MongoClient(uri);
+
+const updateOrCreate = async (collectionName,payload)=> {
+        try {
+          await client.connect();
+          const db=client.db('sink_db');
+          const collection = db.collection(collectionName);
+          const query = { id: payload.id };
+          const update = { $set: payload};
+          const options = { upsert: true };
+          const result= await collection.updateOne(query, update, options);
+          console.log(result);
+        } catch(error) {
+          console.log("error: ", error)
+        }
+      };
+
+const deleteDoc = async (collectionName,payload)=> {
+        try {
+            await client.connect();
+            const db=client.db('sink_db');
+            const query = { id: payload.id };
+            const collection = db.collection(collectionName);
+            const result = await collection.deleteOne(query);
+            console.log(result);
+        } catch(error) {
+            console.log("error: ", error)
+          }
+      }
+
+module.exports={
+    updateOrCreate,
+    deleteDoc
+}