steveo - npm Package Compare versions

Comparing version 1.0.6 to 1.0.7

lib/helpers/kafka.js

@@ -11,10 +11,8 @@ 'use strict';
 
-var _uuid = require('uuid');
-
-var _uuid2 = _interopRequireDefault(_uuid);
-
 function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }
 
 var KafkaClient = function KafkaClient(params) {
-  var producer = new _noKafka2.default.Producer({
+  var admin = new _noKafka2.default.GroupAdmin({
+    groupId: params.kafkaGroupId,
+    clientId: params.clientId,
     connectionString: params.kafkaConnection,
@@ -26,3 +24,3 @@ codec: params.kafkaCodec
     groupId: params.kafkaGroupId,
-    clientId: params.clientId || _uuid2.default.v4(),
+    clientId: params.clientId,
     connectionString: params.kafkaConnection,
@@ -33,8 +31,13 @@ codec: params.kafkaCodec,
 
+  var producer = new _noKafka2.default.Producer({
+    connectionString: params.kafkaConnection,
+    codec: params.kafkaCodec
+  });
+
   return {
     producer: producer,
-    consumer: consumer
+    consumer: consumer,
+    admin: admin
   };
 };
-
 exports.default = KafkaClient;

lib/index.js

@@ -14,2 +14,6 @@ 'use strict';
 
+var _nullLogger = require('null-logger');
+
+var _nullLogger2 = _interopRequireDefault(_nullLogger);
+
 var _task = require('./task');
@@ -27,10 +31,21 @@
 
+var _admin = require('./admin');
+
+var _admin2 = _interopRequireDefault(_admin);
+
 function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }
 
-var Steveo = function Steveo(env, logger) {
+var Steveo = function Steveo(config) {
+  var logger = arguments.length > 1 && arguments[1] !== undefined ? arguments[1] : _nullLogger2.default;
+
   var registeredTopics = {};
-  var registry = (0, _registry2.default)(registeredTopics);
-  var runner = (0, _runner2.default)(env, registry, logger);
+  var registry = (0, _registry2.default)(registeredTopics, config.publishCallback);
+  var task = function task() {
+    var runner = (0, _runner2.default)(config, registeredTopics, logger);
+    return (0, _task2.default)(registry, runner, logger);
+  };
+
   return {
-    task: (0, _task2.default)(registry, runner, logger)
+    task: task,
+    lag: (0, _admin2.default)(config).lag
   };
@@ -37,0 +52,0 @@ };

lib/registry.js

@@ -7,23 +7,84 @@ 'use strict';
 
+var _constants = require('./constants');
 
-var Registry = function Registry(registeredTasks) {
-  var addNewTask = function addNewTask(task, runner) {
-    registeredTasks[task.topic] = task; // eslint-disable-line
-    // call initialize consumer
-    var topics = Object.keys(registeredTasks);
-    runner.initializeConsumer(topics);
-  };
+var _constants2 = _interopRequireDefault(_constants);
 
-  var removeTask = function removeTask(task, runner) {
-    delete registeredTasks[task.topic]; // eslint-disable-line
-    // call initialize consumer
-    var topics = Object.keys(registeredTasks);
-    runner.initializeConsumer(topics);
-  };
+function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }
 
+function _asyncToGenerator(fn) { return function () { var gen = fn.apply(this, arguments); return new Promise(function (resolve, reject) { function step(key, arg) { try { var info = gen[key](arg); var value = info.value; } catch (error) { reject(error); return; } if (info.done) { resolve(value); } else { return Promise.resolve(value).then(function (value) { step("next", value); }, function (err) { step("throw", err); }); } } return step("next"); }); }; }
+
+var Registry = function Registry(registeredTasks, publishCallbacks) {
+  var addNewTask = function () {
+    var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee(task, runner) {
+      return regeneratorRuntime.wrap(function _callee$(_context) {
+        while (1) {
+          switch (_context.prev = _context.next) {
+            case 0:
+              registeredTasks[task.topic] = task; // eslint-disable-line
+              _context.next = 3;
+              return runner.initializeGroupAdmin();
+
+            case 3:
+              _context.next = 5;
+              return runner.initializeProducer();
+
+            case 5:
+              _context.next = 7;
+              return runner.initializeConsumer(Object.keys(registeredTasks));
+
+            case 7:
+            case 'end':
+              return _context.stop();
+          }
+        }
+      }, _callee, undefined);
+    }));
+
+    return function addNewTask(_x, _x2) {
+      return _ref.apply(this, arguments);
+    };
+  }();
+
+  var removeTask = function () {
+    var _ref2 = _asyncToGenerator(regeneratorRuntime.mark(function _callee2(task, runner) {
+      var topics;
+      return regeneratorRuntime.wrap(function _callee2$(_context2) {
+        while (1) {
+          switch (_context2.prev = _context2.next) {
+            case 0:
+              delete registeredTasks[task.topic]; // eslint-disable-line
+              // call initialize consumer
+              topics = Object.keys(registeredTasks);
+              _context2.next = 4;
+              return runner.initializeProducer();
+
+            case 4:
+              _context2.next = 6;
+              return runner.initializeGroupAdmin();
+
+            case 6:
+              _context2.next = 8;
+              return runner.initializeConsumer(topics);
+
+            case 8:
+            case 'end':
+              return _context2.stop();
+          }
+        }
+      }, _callee2, undefined);
+    }));
+
+    return function removeTask(_x3, _x4) {
+      return _ref2.apply(this, arguments);
+    };
+  }();
+
   return {
     addNewTask: addNewTask,
-    removeTask: removeTask
+    removeTask: removeTask,
+    successCallback: publishCallbacks && publishCallbacks.success ? publishCallbacks.success : _constants2.default.NOOP,
+    failureCallback: publishCallbacks && publishCallbacks.failure ? publishCallbacks.failure : _constants2.default.NOOP
   };
 };
+
 exports.default = Registry;

lib/runner.js

@@ -11,2 +11,6 @@ 'use strict';
 
+var _noKafka = require('no-kafka');
+
+var _noKafka2 = _interopRequireDefault(_noKafka);
+
 var _kafka = require('./helpers/kafka');
@@ -20,52 +24,65 @@
 
-var Runner = function Runner(env, registry, logger) {
+var Runner = function Runner(config, registry, logger) {
   var kafkaClient = (0, _kafka2.default)({
-    kafkaConnection: env.KAFKA_CONNECTION,
-    kafkaCodec: env.KAFKA_CODEC,
-    clientId: env.CLIENT_ID,
+    kafkaConnection: config.kafkaConnection,
+    kafkaCodec: config.kafkaCodec,
+    clientId: config.clientId,
     logger: {
-      logLevel: env.LOG_LEVEL
+      logLevel: config.logLevel
     },
-    kafkaGroupId: env.KAFKA_GROUP_ID
+    kafkaGroupId: config.kafkaGroupId
   });
 
-  var receive = function () {
-    var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee(payload, topic) {
-      var task;
-      return regeneratorRuntime.wrap(function _callee$(_context) {
-        while (1) {
-          switch (_context.prev = _context.next) {
-            case 0:
-              logger.info('Payload: ', JSON.stringify(payload, null, 2), 'received on topic:', topic);
-              task = registry[topic];
-              _context.prev = 2;
-              _context.next = 5;
-              return task.subscribe(payload);
+  var receive = function receive(messageSet, topic, partition) {
+    return Promise.all(messageSet.map(function () {
+      var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee(m) {
+        var task;
+        return regeneratorRuntime.wrap(function _callee$(_context) {
+          while (1) {
+            switch (_context.prev = _context.next) {
+              case 0:
+                _context.prev = 0;
 
-            case 5:
-              _context.next = 10;
-              break;
+                logger.info('\n        *****CONSUME********\n        topic:- ' + topic + '\n        ********************\n        payload:- ' + m.message.value + '\n        ********************\n      ');
+                // commit offset
+                _context.next = 4;
+                return kafkaClient.consumer.commitOffset({ topic: topic, partition: partition, offset: m.offset, metadata: 'optional' });
 
-            case 7:
-              _context.prev = 7;
-              _context.t0 = _context['catch'](2);
+              case 4:
+                task = registry[topic];
+                _context.next = 7;
+                return task.subscribe(JSON.parse(m.message.value.toString('utf8')));
 
-              logger.error('Error while executing consumer callback ', _context.t0);
+              case 7:
+                _context.next = 12;
+                break;
 
-            case 10:
-            case 'end':
-              return _context.stop();
+              case 9:
+                _context.prev = 9;
+                _context.t0 = _context['catch'](0);
+
+                logger.error('Error while executing consumer callback ', _context.t0);
+
+              case 12:
+              case 'end':
+                return _context.stop();
+            }
           }
-        }
-      }, _callee, undefined, [[2, 7]]);
-    }));
+        }, _callee, undefined, [[0, 9]]);
+      }));
 
-    return function receive(_x, _x2) {
-      return _ref.apply(this, arguments);
-    };
-  }();
+      return function (_x) {
+        return _ref.apply(this, arguments);
+      };
+    }()));
+  };
 
   var initializeConsumer = function initializeConsumer(subscriptions) {
     logger.info('initializing consumer', subscriptions);
+
     return kafkaClient.consumer.init([{
+      strategy: new _noKafka2.default.WeightedRoundRobinAssignmentStrategy(),
+      metadata: {
+        weight: 4
+      },
       subscriptions: subscriptions,
@@ -76,2 +93,10 @@ handler: receive
 
+  var initializeGroupAdmin = function initializeGroupAdmin() {
+    return kafkaClient.admin.init();
+  };
+
+  var initializeProducer = function initializeProducer() {
+    return kafkaClient.producer.init();
+  };
+
   var producerPayload = function producerPayload(msg, topic) {
@@ -96,29 +121,27 @@ var timestamp = (0, _moment2.default)().unix();
               data = producerPayload(payload, topic);
-
-              logger.info('*****', data);
               sendParams = {
                 retries: {
-                  attempts: env.KAFKA_SEND_ATTEMPTS,
+                  attempts: config.kafkaSendAttempts,
                   delay: {
-                    min: env.KAFKA_SEND_DELAY_MIN,
-                    max: env.KAFKA_SEND_DELAY_MAX
+                    min: config.kafkaSendDelayMin,
+                    max: config.kafkaSendDelayMax
                   }
                 }
               };
-              _context2.prev = 4;
-              _context2.next = 7;
+              _context2.prev = 3;
+              _context2.next = 6;
               return kafkaClient.producer.send(data, sendParams);
 
-            case 7:
+            case 6:
               logger.info('\n        *****PRODUCE********\n        topic:- ' + topic + '\n        ********************\n        payload:- ' + JSON.stringify(payload) + '\n        ********************\n      ');
-              _context2.next = 13;
+              _context2.next = 12;
               break;
 
-            case 10:
-              _context2.prev = 10;
-              _context2.t0 = _context2['catch'](4);
+            case 9:
+              _context2.prev = 9;
+              _context2.t0 = _context2['catch'](3);
 
               logger.error('Error while sending payload:', JSON.stringify(payload, null, 2), 'topic :', topic, 'Error :', _context2.t0);
 
-            case 13:
+            case 12:
             case 'end':
@@ -128,6 +151,6 @@ return _context2.stop();
         }
-      }, _callee2, undefined, [[4, 10]]);
+      }, _callee2, undefined, [[3, 9]]);
     }));
 
-    return function send(_x3, _x4) {
+    return function send(_x2, _x3) {
       return _ref2.apply(this, arguments);
@@ -137,2 +160,9 @@ };
 
+  var fetchConsumerLag = function fetchConsumerLag(groupId, topicName, partitions) {
+    return kafkaClient.admin.fetchConsumerLag(groupId, [{
+      topicName: topicName,
+      partitions: partitions
+    }]);
+  };
+
   return {
@@ -142,3 +172,6 @@ send: send,
     kafkaClient: kafkaClient,
-    initializeConsumer: initializeConsumer
+    initializeConsumer: initializeConsumer,
+    initializeGroupAdmin: initializeGroupAdmin,
+    initializeProducer: initializeProducer,
+    fetchConsumerLag: fetchConsumerLag
   };
@@ -145,0 +178,0 @@ };

lib/task.js

@@ -25,14 +25,5 @@ 'use strict';
 
-  var define = function define(topicName, callBack) {
-    topic = topicName;
-    subscribeCallback = callBack;
-    var task = {
-      topic: topic,
-      subscribe: subscribeCallback
-    };
-    registry.addNewTask(task, runner);
-  };
-
-  var publish = function () {
-    var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee(payload) {
+  var define = function () {
+    var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee(topicName, callBack) {
+      var task;
       return regeneratorRuntime.wrap(function _callee$(_context) {
@@ -42,9 +33,12 @@ while (1) {
             case 0:
-              _context.next = 2;
-              return runner.send(topic, payload);
+              topic = topicName;
+              subscribeCallback = callBack;
+              task = {
+                topic: topic,
+                subscribe: subscribeCallback
+              };
+              _context.next = 5;
+              return registry.addNewTask(task, runner);
 
-            case 2:
-              logger.info(topic, payload);
-
-            case 3:
+            case 5:
             case 'end':
@@ -57,3 +51,3 @@ return _context.stop();
 
-    return function publish(_x) {
+    return function define(_x, _x2) {
       return _ref.apply(this, arguments);
@@ -63,6 +57,43 @@ };
 
+  var publish = function () {
+    var _ref2 = _asyncToGenerator(regeneratorRuntime.mark(function _callee2(payload) {
+      return regeneratorRuntime.wrap(function _callee2$(_context2) {
+        while (1) {
+          switch (_context2.prev = _context2.next) {
+            case 0:
+              _context2.prev = 0;
+              _context2.next = 3;
+              return runner.send(topic, payload);
+
+            case 3:
+              logger.info('FROM PRODUCER', topic, payload);
+              registry.successCallback(topic, payload);
+              _context2.next = 11;
+              break;
+
+            case 7:
+              _context2.prev = 7;
+              _context2.t0 = _context2['catch'](0);
+
+              registry.failureCallback(topic, payload);
+              throw _context2.t0;
+
+            case 11:
+            case 'end':
+              return _context2.stop();
+          }
+        }
+      }, _callee2, undefined, [[0, 7]]);
+    }));
+
+    return function publish(_x3) {
+      return _ref2.apply(this, arguments);
+    };
+  }();
+
   return {
     define: define,
     publish: publish,
-    subscribe: subscribe
+    subscribe: subscribe,
+    runner: runner
   };
@@ -69,0 +100,0 @@ };

package.json

 {
   "name": "steveo",
-  "version": "1.0.6",
+  "version": "1.0.7",
   "description": "A Task Manager Library",
@@ -28,3 +28,4 @@ "main": "lib/index.js",
       "types",
-      "src/constants"
+      "src/constants",
+      "src/admin.js"
     ]
@@ -37,2 +38,3 @@ },
     "no-kafka": "^3.1.1",
+    "null-logger": "^1.0.0",
     "uuid": "^3.0.1"
@@ -39,0 +41,0 @@ },

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

31567

increased by30.06%

Number of package files

11

increased by10%

Lines of code

428

increased by65.25%

Worsened metrics

Dependency count

6

increased by20%

Dependency changes

• + Addednull-logger@^1.0.0

• + Addednull-logger@1.0.0(transitive)