Mysql DB

Author: piyal-ict-mbstu

.env

@@ -1,5 +1,9 @@
-
+#source database config
 MYSQL_HOST="localhost"
 MYSQL_PORT="3306"
 MYSQL_USER="root"
-MYSQL_PASSWORD="rootPwd@2022"
+MYSQL_PASSWORD="rootPwd@2022"
+
+#sink data config
+MONGO_DB_URL="mongodb://localhost:27017"
+MONGO_DB_COLLECTION_NAME="nise"

Producer/Producer.js

@@ -0,0 +1,3 @@
+const mysqlDBSinker = require('./source-connector/mysqlSourceConnector');
+mysqlDBSinker.then(() => console.log('Waiting for database vents...'))
+    .catch(console.error);

Source-Connector/mysql-source.js

@@ -0,0 +1,5 @@
+module.exports = {
+    EVENT_TYPE_INSERT: "INSERT",
+    EVENT_TYPE_UPDATE: "UPDATE",
+    EVENT_TYPE_DELETE: "DELETE",
+};

app.js

@@ -0,0 +1,21 @@
+module.exports = {
+    /**
+     * Default: false
+     * Pass true to only emit binlog events that occur after ZongJi's instantiation.
+     * Must be used in start() method for effect.
+     */
+    startAtEnd: true,
+    /**
+     * Databases and tables to include (Only for row events).
+     * Use database names as the key and pass an array of table names or true (for the entire database)
+     * Example: { 'my_database': ['allow_table', 'another_table'], 'another_db': true }
+     */
+    includeSchema: {},
+    /**
+     * Object describing which databases and tables to exclude (Same format as includeSchema)
+     * Example: { 'other_db': ['disallowed_table'], 'ex_db': true }
+     */
+    excludeSchema: {
+        mysql: true
+    }
+}

common/common.js

@@ -1,20 +1,14 @@
 {
-  "name": "CDC",
+  "name": "cdc",
   "version": "1.0.1",
-  "description": "Change data chapture",
+  "description": "Change data capture",
   "main": "server.js",
   "repository": "https://github.com/tasmidur/cdc.git",
   "author": "Md Tasmidur Rahman",
   "license": "MIT",
   "private": false,
   "scripts": {
-    "start": "node Source-Connector/mysql-source.js",
-    "source": "node Source-Connector/mysql-source.js",
-    "sink": "node consumer/sink-connector.js",
-    "serve": "vue-cli-service serve",
-    "build": "vue-cli-service build",
-    "lint": "vue-cli-service lint",
-    "test:unit": "vue-cli-service test:unit --watch"
+    "start": "node app.js"
   },
   "dependencies": {
     "@rodrigogs/mysql-events": "^0.6.0",

common/event-config.js

@@ -0,0 +1,39 @@
+const {MongoClient} = require("mongodb");
+require('dotenv').config();
+
+const uri = process.env.MONGO_DB_URL;
+const collectionName=process.env.MONGO_DB_COLLECTION_NAME
+const client = new MongoClient(uri);
+
+const updateOrCreate = async (documentName, payload) => {
+    try {
+        await client.connect();
+        const db = client.db(collectionName);
+        const collection = db.collection(documentName);
+        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 (documentName, payload) => {
+    try {
+        await client.connect();
+        const db = client.db(collectionName);
+        const query = {id: payload.id};
+        const collection = db.collection(documentName);
+        const result = await collection.deleteOne(query);
+        console.log(result);
+    } catch (error) {
+        console.log("error: ", error)
+    }
+}
+
+module.exports = {
+    updateOrCreate,
+    deleteDoc
+}