added queue support for insert update and delete operation

Author: hkulekci

README.md

@@ -4,8 +4,8 @@
 
 ```
 docker-compose up elasticsearch kibana
-docker-compose up mysql
-docker-compose up app
+docker-compose up mysql redis
+docker-compose up app listener
 ```
 
 ## Database Structure and Initialization
@@ -21,6 +21,12 @@ Port : 33060
 
 And import `data/sample_data.sql` file for our sample data.
 
+Redis configuration :
+
+```
+Host: redis
+```
+
 
 ## Elastic Integration
 

docker-compose.yml

@@ -59,3 +59,17 @@ services:
             - elasticsearch
             - mysql
             - redis
+
+    listener:
+        image: "node:8"
+        user: "node"
+        working_dir: /home/node/app
+        environment:
+            - NODE_ENV=production
+        volumes:
+            - ./:/home/node/app
+        command: "npm run listen"
+        links:
+            - elasticsearch
+            - mysql
+            - redis

libraries/redis-client.js

@@ -2,6 +2,10 @@ var redis = require('redis');
   client = redis.createClient({'host': 'redis'});
 
 var RedisClient = {
+  getClient: function() {
+    return client;
+  },
+
   push: function(key, value, callback) {
     client.lpush(key, value, function(err, result) {
       callback(err, result);

package-lock.json

@@ -3,7 +3,8 @@
   "version": "0.0.0",
   "private": true,
   "scripts": {
-    "start": "nodemon -e js,twig,css ./bin/www"
+    "start": "nodemon -e js,twig,css ./bin/www",
+    "listen": "nodemon -e queue-listener.js ./queue-listener.js"
   },
   "dependencies": {
     "async": "^2.5.0",
@@ -16,11 +17,12 @@
     "express": "~4.15.5",
     "less-middleware": "~2.2.1",
     "morgan": "~1.9.0",
-    "mysql": "^2.14.1",
+    "mysql": "^2.15.0",
     "node-datetime": "^2.0.3",
     "nodemon": "^1.12.1",
     "redis": "^2.8.0",
     "serve-favicon": "~2.4.5",
+    "simple-redis-queue": "git+ssh://[email protected]/hkulekci/node-simple-redis-queue.git",
     "twig": "~0.10.3"
   }
 }

package.json

@@ -0,0 +1,77 @@
+#!/usr/bin/env node
+
+require('dotenv').config();
+var productService = require('./services/productService');
+var categoryService = require('./services/categoryService');
+var productSearchService = require('./services/productSearchService');
+var redisClient = require('./libraries/redis-client');
+var RedisQueue = require("simple-redis-queue");
+var waterfall = require('async/waterfall');
+
+var pop_queue = new RedisQueue(redisClient.getClient());
+
+var upsertOperation = function(productId, functionCallback) {
+  waterfall([
+    // Insert Operation
+    function(waterfallCallback) { // Getting Product Category From DB (Preparing for Elasticsearch)
+      var result = {};
+      categoryService.getProductCategories(productId, function(err, productCategories) {
+        if (err) {}
+        result.productCategories = productCategories;
+        waterfallCallback(false, result);
+      });
+    },
+    function(result, waterfallCallback) { // Getting Product Data From MySQL DB (Preparing for Elasticsearch)
+      productService.getRecord(productId, function(err, product) {
+        if (err) {}
+        result.product = product;
+        waterfallCallback(false, result);
+      });
+    },
+    function(result, waterfallCallback) { // Saving Product to Elasticsearch
+      var product = result.product;
+      product.categories = result.productCategories;
+
+      productSearchService.insert(product, function() {
+        waterfallCallback(false, product);
+      });
+    },
+    function(err, result) {
+      functionCallback(err, result);
+    }
+  ]);
+};
+
+var deleteOperation = function(productId, functionCallback) {
+    // Delete Operation
+    productSearchService.delete(productId, function(err, result) {
+      functionCallback(false);
+    });
+};
+
+
+/***** Redis Message Queue Listener ******/
+pop_queue.on("message", function (queueName, payload) {
+  var messageData = JSON.parse(payload);
+  if (messageData.action == 'update' || messageData.action == 'insert') {
+    upsertOperation(messageData.productId, function() {
+      console.log('[' + queueName + '] - Processed! - ' + payload);
+      pop_queue.next("product_updates");
+    });
+  } else if (messageData.action == 'delete') {
+    deleteOperation(messageData.productId, function() {
+      console.log('[' + queueName + '] - Processed! - ' + payload);
+      pop_queue.next("product_updates");
+    });
+  } else {
+    console.log('[' + queueName + '] - Not Processed! - ' + payload);
+    pop_queue.next("product_updates");
+  }
+});
+
+// Listen for errors
+pop_queue.on("error", function (error) {
+    console.log("pop_queue Error : " + error);
+});
+
+pop_queue.next("product_updates");

queue-listener.js

@@ -6,6 +6,9 @@ var waterfall = require('async/waterfall');
 var productSearchService = require('../services/productSearchService');
 var redisClient = require('../libraries/redis-client');
 var md5 = require('blueimp-md5');
+var RedisQueue = require("simple-redis-queue");
+
+var push_queue = new RedisQueue(redisClient.getClient());
 
 /* GET products listing. */
 router.get('/', function(req, res, next) {
@@ -80,32 +83,15 @@ router.post('/new', function(req, res, next) {
           })
         });
       },
-      function(result, waterfallCallback) { // Getting Product Category From DB (Preparing for Elasticsearch)
-        categoryService.getProductCategories(result.insertId, function(err, productCategories) {
-          if (err) {}
-          result.productCategories = productCategories;
-          waterfallCallback(false, result);
-        });
-      },
-      function(result, waterfallCallback) { // Getting Product Data From MySQL DB (Preparing for Elasticsearch)
-        productService.getRecord(result.insertId, function(err, product) {
-          if (err) {}
-          result.product = product;
+      function(result, waterfallCallback) {
+        push_queue.push('product_updates', {'action':'insert', 'productId': result.insertId}, function(err, res) {
           waterfallCallback(false, result);
         });
-      },
-      function(result, waterfallCallback) { // Saving Product to Elasticsearch
-        var product = result.product;
-        product.categories = result.productCategories;
-
-        productSearchService.insert(product, function() {
-          waterfallCallback(false, product);
-        });
       }
     ],
-    function(err, product) {
+    function(err, result) {
       if (err) {}
-      res.redirect('/product/id/'+product.id);
+      res.redirect('/product/id/'+result.insertId);
     }
   );
 });
@@ -122,8 +108,7 @@ router.get('/:id/delete', function(req, res, next) {
         });
       },
       function(waterfallCallback) {
-        productSearchService.delete(params.id, function(err, result) {
-          //TODO: check error status
+        push_queue.push('product_updates', {'action':'delete', 'productId': params.id}, function(err, res) {
           waterfallCallback(false);
         });
       }
@@ -210,33 +195,15 @@ router.post('/:id/edit', function(req, res, next) {
           })
         });      
       },
-      function(waterfallCallback) { // Getting Product Categories Data from MySQL (Preparing for Elasticsearch)
-        var result = {};
-        categoryService.getProductCategories(params.id, function(err, productCategories) {
-          if (err) {}
-          result.productCategories = productCategories;
-          waterfallCallback(false, result);
-        });
-      },
-      function(result, waterfallCallback) { // Getting Product Data from MySQL  (Preparing for Elasticsearch)
-        productService.getRecord(params.id, function(err, product) {
-          if (err) {}
-          result.product = product;
-          waterfallCallback(false, result);
-        });
-      },
-      function(result, waterfallCallback) { // Saving Data to Elasticsearch
-        var product = result.product;
-        product.categories = result.productCategories;
-
-        productSearchService.insert(product, function() {
-          waterfallCallback(false, product);
+      function(waterfallCallback) {
+        push_queue.push('product_updates', {'action':'update', 'productId': queryParams.id}, function(err, res) {
+          waterfallCallback(false);
         });
       }
     ],
     function(err, product) {
       if (err) {}
-      res.redirect('/product/id/' + product.id);
+      res.redirect('/product/id/' + queryParams.id);
       return;
     }
     );