Adhidarma Hadiwinoto · Adhidarma Hadiwinoto
Showing 4 changed files Side-by-side Diff
gateway/pull.js
gateway/resend-delay.js
package.json
resend-delay.js
@@ -0,0 +1,148 @@
+"use strict";
+
+const request = require('request');
+
+const config = require('../config');
+const logger = require('../logger');
+const matrix = require('../matrix');
+
+var partner;
+
+function init(options) {
+    partner = options.partner;
+
+    initMatrix();
+
+    setInterval(pullTask, config.pull_interval_ms || 1000);
+}
+
+function pullTask() {
+    let options = {
+        url: config.pull_url.task.replace('<CORE_APIKEY>', config.core_apikey),
+        qs: {
+            handler: config.handler_name,
+            products: config.products.join(',')
+        }
+    }
+
+    request(options, function(error, response, body) {
+        if (error) {
+            if (matrix.core_is_healthy) {
+                logger.warn('Error pulling task from CORE', {error: error});
+            }
+            matrix.core_is_healthy = false;
+            return;
+        }
+
+        if (response.statusCode != 200) {
+            if (matrix.core_is_healthy) {
+                logger.warn('CORE http response status code for pull task is not 200', {http_response_status: response.statusCode});
+            }
+            matrix.core_is_healthy = false;
+            return;
+        }
+
+        if (!matrix.core_is_healthy) {
+            logger.verbose('CORE is healthy');
+        }
+        matrix.core_is_healthy = true;
+
+        if (body == 'NONE') {
+            return;
+        }
+
+        forwardCoreTaskToPartner(body);
+    });
+}
+
+function forwardCoreTaskToPartner(coreMessage) {
+    let task;
+
+    try {
+        task = JSON.parse(coreMessage);
+    }
+    catch(e) {
+        logger.warn('Exception on parsing CORE pull task response', {coreMessage: coreMessage, error: e});
+    }
+
+    task.remote_product = getRemoteProduct(task.product);
+
+    partner.buy(task);
+}
+
+function report(trx_id, rc, message, sn) {
+    let options = {
+        url: config.pull_url.report.replace('<CORE_APIKEY>', config.core_apikey),
+        qs: {
+            trx_id: trx_id,
+            rc: rc,
+            message: message,
+            handler: config.handler_name
+        }
+    }
+
+    if (sn) {
+        options.qs.sn = sn;
+    }
+
+    request(options, function(error, response, body) {
+        if (error) {
+            logger.warn('Error reporting to CORE', {error: error});
+        }
+        else if (response.statusCode != 200) {
+            logger.warn('CORE http response status is not 200', {requestOptions: options, http_response_status: response.statusCode});
+        }
+        else {
+            logger.verbose('Report has been sent to CORE', {requestOptions: options});
+        }
+    });
+}
+
+function resendReport(trx_id, rc, message, sn) {
+    let sleepBeforeResend = 1000;
+    logger.verbose('Resend report to CORE in ' + sleepBeforeResend + 'ms')
+
+    setTimeout(
+        function() {
+            report(trx_id, rc, message, sn);
+        },
+        sleepBeforeResend
+    )
+}
+
+function isPaused() {
+    return matrix.paused;
+}
+
+function pause() {
+    matrix.paused = true;
+}
+
+function resume() {
+    matrix.pause = false;
+}
+
+function initMatrix() {
+    if (!matrix) {
+        matrix = {};
+    }
+
+    matrix.counter = {
+        trx: 0
+    }
+}
+
+function incrementCounterTrx() {
+    matrix.counter.trx++;
+}
+
+function getRemoteProduct(product) {
+    let remoteProduct = config.remote_products[product];
+    return remoteProduct || product;
+}
+
+exports.init = init;
+exports.isPaused = isPaused;
+exports.pause = pause;
+exports.resume = resume;
+exports.report = report;
@@ -0,0 +1,87 @@
+"use strict";
+
+var LRU = require('lru-cache');
+
+const config = require('./config');
+const logger = require('./logger');
+
+var topupRequest;
+var resendHandlers = LRU({max: 2000, maxAge: 1000 * 3600 * 36});
+
+function init(options) {
+    if (options && options.request) {
+        request = options.request;
+    } else {
+        logger.warn('Undefined options.request, terminating....');
+        process.exit(1);
+    }
+}
+
+function cancel(task) {
+    var requestId;
+    if (typeof task === 'string') {
+        requestId = task;
+    } else {
+        requestId = task.requestId;
+    }
+
+    if (!requestId) {
+        return;
+    }
+
+    var oldHandler = resendHandlers.get(requestId);
+    if (!oldHandler) {
+        return;
+    }
+
+    logger.verbose('Canceling resend delay', {task: oldHandler.task});
+
+    try {
+        if (oldHandler.handler) {
+            clearTimeout(oldHandler.handler);
+        }
+    }
+    catch(e) {};
+
+    try {
+        resendHandlers.del(requestId);
+    }
+    catch(e) {};
+}
+
+function register(task) {
+    if (!task.requestId) {
+        logger.warn('Invalid task on resendDelay')
+        return;
+    }
+
+    if (!config || !config.auto_resend_delay || !Number(config.auto_resend_delay.delay_ms) || !Number(config.auto_resend_delay.resend_max)) {
+        return;
+    }
+
+    var retry = config.auto_resend_delay.resend_max;
+    var oldHandler = resendHandlers.get(task.requestId);
+    if (oldHandler) {
+        retry = oldHandler.retry - 1;
+        cancel(task);
+    }
+
+    if (retry <= 0) {
+        logger.verbose('Resend delay retry exceeded', {task: task});
+        cancel(task);
+        return;
+    }
+
+    logger.verbose('Registering resend delay task request', {task: task, delay_ms: config.auto_resend_delay.delay_ms, retry: retry});
+    var handlerData = {
+        handler: setTimeout(request, config.auto_resend_delay.delay_ms, task),
+        task: task,
+        retry: retry
+    }
+
+    resendHandlers.set(task.requestId, handlerData);
+}
+
+exports.init = init;
+exports.cancel = cancel;
+exports.register = register;
 {
   "name": "komodo-sdk",
-  "version": "1.5.1",
+  "version": "1.6.0",
   "description": "SDK for Komodo",
   "main": "index.js",
   "scripts": {
...	...	@@ -0,0 +1,148 @@
	1	+"use strict";
	2	+
	3	+const request = require('request');
	4	+
	5	+const config = require('../config');
	6	+const logger = require('../logger');
	7	+const matrix = require('../matrix');
	8	+
	9	+var partner;
	10	+
	11	+function init(options) {
	12	+ partner = options.partner;
	13	+
	14	+ initMatrix();
	15	+
	16	+ setInterval(pullTask, config.pull_interval_ms \|\| 1000);
	17	+}
	18	+
	19	+function pullTask() {
	20	+ let options = {
	21	+ url: config.pull_url.task.replace('<CORE_APIKEY>', config.core_apikey),
	22	+ qs: {
	23	+ handler: config.handler_name,
	24	+ products: config.products.join(',')
	25	+ }
	26	+ }
	27	+
	28	+ request(options, function(error, response, body) {
	29	+ if (error) {
	30	+ if (matrix.core_is_healthy) {
	31	+ logger.warn('Error pulling task from CORE', {error: error});
	32	+ }
	33	+ matrix.core_is_healthy = false;
	34	+ return;
	35	+ }
	36	+
	37	+ if (response.statusCode != 200) {
	38	+ if (matrix.core_is_healthy) {
	39	+ logger.warn('CORE http response status code for pull task is not 200', {http_response_status: response.statusCode});
	40	+ }
	41	+ matrix.core_is_healthy = false;
	42	+ return;
	43	+ }
	44	+
	45	+ if (!matrix.core_is_healthy) {
	46	+ logger.verbose('CORE is healthy');
	47	+ }
	48	+ matrix.core_is_healthy = true;
	49	+
	50	+ if (body == 'NONE') {
	51	+ return;
	52	+ }
	53	+
	54	+ forwardCoreTaskToPartner(body);
	55	+ });
	56	+}
	57	+
	58	+function forwardCoreTaskToPartner(coreMessage) {
	59	+ let task;
	60	+
	61	+ try {
	62	+ task = JSON.parse(coreMessage);
	63	+ }
	64	+ catch(e) {
	65	+ logger.warn('Exception on parsing CORE pull task response', {coreMessage: coreMessage, error: e});
	66	+ }
	67	+
	68	+ task.remote_product = getRemoteProduct(task.product);
	69	+
	70	+ partner.buy(task);
	71	+}
	72	+
	73	+function report(trx_id, rc, message, sn) {
	74	+ let options = {
	75	+ url: config.pull_url.report.replace('<CORE_APIKEY>', config.core_apikey),
	76	+ qs: {
	77	+ trx_id: trx_id,
	78	+ rc: rc,
	79	+ message: message,
	80	+ handler: config.handler_name
	81	+ }
	82	+ }
	83	+
	84	+ if (sn) {
	85	+ options.qs.sn = sn;
	86	+ }
	87	+
	88	+ request(options, function(error, response, body) {
	89	+ if (error) {
	90	+ logger.warn('Error reporting to CORE', {error: error});
	91	+ }
	92	+ else if (response.statusCode != 200) {
	93	+ logger.warn('CORE http response status is not 200', {requestOptions: options, http_response_status: response.statusCode});
	94	+ }
	95	+ else {
	96	+ logger.verbose('Report has been sent to CORE', {requestOptions: options});
	97	+ }
	98	+ });
	99	+}
	100	+
	101	+function resendReport(trx_id, rc, message, sn) {
	102	+ let sleepBeforeResend = 1000;
	103	+ logger.verbose('Resend report to CORE in ' + sleepBeforeResend + 'ms')
	104	+
	105	+ setTimeout(
	106	+ function() {
	107	+ report(trx_id, rc, message, sn);
	108	+ },
	109	+ sleepBeforeResend
	110	+ )
	111	+}
	112	+
	113	+function isPaused() {
	114	+ return matrix.paused;
	115	+}
	116	+
	117	+function pause() {
	118	+ matrix.paused = true;
	119	+}
	120	+
	121	+function resume() {
	122	+ matrix.pause = false;
	123	+}
	124	+
	125	+function initMatrix() {
	126	+ if (!matrix) {
	127	+ matrix = {};
	128	+ }
	129	+
	130	+ matrix.counter = {
	131	+ trx: 0
	132	+ }
	133	+}
	134	+
	135	+function incrementCounterTrx() {
	136	+ matrix.counter.trx++;
	137	+}
	138	+
	139	+function getRemoteProduct(product) {
	140	+ let remoteProduct = config.remote_products[product];
	141	+ return remoteProduct \|\| product;
	142	+}
	143	+
	144	+exports.init = init;
	145	+exports.isPaused = isPaused;
	146	+exports.pause = pause;
	147	+exports.resume = resume;
	148	+exports.report = report;
...	...	@@ -0,0 +1,87 @@
	1	+"use strict";
	2	+
	3	+var LRU = require('lru-cache');
	4	+
	5	+const config = require('./config');
	6	+const logger = require('./logger');
	7	+
	8	+var topupRequest;
	9	+var resendHandlers = LRU({max: 2000, maxAge: 1000 * 3600 * 36});
	10	+
	11	+function init(options) {
	12	+ if (options && options.request) {
	13	+ request = options.request;
	14	+ } else {
	15	+ logger.warn('Undefined options.request, terminating....');
	16	+ process.exit(1);
	17	+ }
	18	+}
	19	+
	20	+function cancel(task) {
	21	+ var requestId;
	22	+ if (typeof task === 'string') {
	23	+ requestId = task;
	24	+ } else {
	25	+ requestId = task.requestId;
	26	+ }
	27	+
	28	+ if (!requestId) {
	29	+ return;
	30	+ }
	31	+
	32	+ var oldHandler = resendHandlers.get(requestId);
	33	+ if (!oldHandler) {
	34	+ return;
	35	+ }
	36	+
	37	+ logger.verbose('Canceling resend delay', {task: oldHandler.task});
	38	+
	39	+ try {
	40	+ if (oldHandler.handler) {
	41	+ clearTimeout(oldHandler.handler);
	42	+ }
	43	+ }
	44	+ catch(e) {};
	45	+
	46	+ try {
	47	+ resendHandlers.del(requestId);
	48	+ }
	49	+ catch(e) {};
	50	+}
	51	+
	52	+function register(task) {
	53	+ if (!task.requestId) {
	54	+ logger.warn('Invalid task on resendDelay')
	55	+ return;
	56	+ }
	57	+
	58	+ if (!config \|\| !config.auto_resend_delay \|\| !Number(config.auto_resend_delay.delay_ms) \|\| !Number(config.auto_resend_delay.resend_max)) {
	59	+ return;
	60	+ }
	61	+
	62	+ var retry = config.auto_resend_delay.resend_max;
	63	+ var oldHandler = resendHandlers.get(task.requestId);
	64	+ if (oldHandler) {
	65	+ retry = oldHandler.retry - 1;
	66	+ cancel(task);
	67	+ }
	68	+
	69	+ if (retry <= 0) {
	70	+ logger.verbose('Resend delay retry exceeded', {task: task});
	71	+ cancel(task);
	72	+ return;
	73	+ }
	74	+
	75	+ logger.verbose('Registering resend delay task request', {task: task, delay_ms: config.auto_resend_delay.delay_ms, retry: retry});
	76	+ var handlerData = {
	77	+ handler: setTimeout(request, config.auto_resend_delay.delay_ms, task),
	78	+ task: task,
	79	+ retry: retry
	80	+ }
	81	+
	82	+ resendHandlers.set(task.requestId, handlerData);
	83	+}
	84	+
	85	+exports.init = init;
	86	+exports.cancel = cancel;
	87	+exports.register = register;
1	1	{
2	2	"name": "komodo-sdk",
3		- "version": "1.5.1",
	3	+ "version": "1.6.0",
4	4	"description": "SDK for Komodo",
5	5	"main": "index.js",
6	6	"scripts": {