steveo - npm Package Compare versions

Comparing version 1.0.4 to 1.0.5

lib/index.js

@@ -7,2 +7,4 @@ 'use strict';
 
+require('babel-polyfill');
+
 var _task = require('./task');
@@ -22,10 +24,11 @@
 
-var Steveo = function Steveo(env, kafkaHost) {
+var Steveo = function Steveo(env, kafkaHost, logger) {
   var registeredTopics = {};
   var registry = (0, _registry2.default)(registeredTopics);
-  var runner = (0, _runner2.default)(env, kafkaHost, registry);
+  var runner = (0, _runner2.default)(env, registry, logger);
   return {
-    task: (0, _task2.default)(registry, runner)
+    task: (0, _task2.default)(registry, runner, logger)
   };
 };
+
 exports.default = Steveo;

lib/registry.js

@@ -1,2 +0,2 @@
-"use strict";
+'use strict';
 
@@ -6,9 +6,17 @@ Object.defineProperty(exports, "__esModule", {
 });
+
+
 var Registry = function Registry(registeredTasks) {
-  var addNewTask = function addNewTask(task) {
+  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 removeTask = function removeTask(task) {
+  var removeTask = function removeTask(task, runner) {
     delete registeredTasks[task.topic]; // eslint-disable-line
+    // call initialize consumer
+    var topics = Object.keys(registeredTasks);
+    runner.initializeConsumer(topics);
   };
@@ -21,3 +29,2 @@
 };
-
 exports.default = Registry;

lib/runner.js

@@ -1,2 +0,2 @@
-"use strict";
+'use strict';
 
@@ -7,9 +7,24 @@ Object.defineProperty(exports, "__esModule", {
 
+var _moment = require('moment');
+
+var _moment2 = _interopRequireDefault(_moment);
+
+var _kafka = require('./helpers/kafka');
+
+var _kafka2 = _interopRequireDefault(_kafka);
+
+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 Runner = function Runner(env, kafkaHost, registry) {
-  var send = function send(topic, payload) {
-    // send messages to kafka
-    console.log(topic, payload);
-  };
+var Runner = function Runner(env, registry, logger) {
+  var kafkaClient = (0, _kafka2.default)({
+    kafkaConnection: env.KAFKA_CONNECTION,
+    kafkaCodec: env.KAFKA_CODEC,
+    clientId: env.CLIENT_ID,
+    logger: {
+      logLevel: env.LOG_LEVEL
+    },
+    kafkaGroupId: env.KAFKA_GROUP_ID
+  });
 
@@ -23,13 +38,24 @@ var receive = function () {
             case 0:
-              // receive messages from kafka
+              logger.info('Payload: ', JSON.stringify(payload, null, 2), 'received on topic:', topic);
               task = registry[topic];
-              _context.next = 3;
+              _context.prev = 2;
+              _context.next = 5;
               return task.subscribe(payload);
 
-            case 3:
-            case "end":
+            case 5:
+              _context.next = 10;
+              break;
+
+            case 7:
+              _context.prev = 7;
+              _context.t0 = _context['catch'](2);
+
+              logger.error('Error while executing consumer callback ', _context.t0);
+
+            case 10:
+            case 'end':
               return _context.stop();
           }
         }
-      }, _callee, undefined);
+      }, _callee, undefined, [[2, 7]]);
     }));
@@ -42,5 +68,73 @@
 
+  var initializeConsumer = function initializeConsumer(subscriptions) {
+    logger.info('initializing consumer', subscriptions);
+    return kafkaClient.consumer.init([{
+      subscriptions: subscriptions,
+      handler: receive
+    }]);
+  };
+
+  var producerPayload = function producerPayload(msg, topic) {
+    var timestamp = (0, _moment2.default)().unix();
+
+    return {
+      timestamp: timestamp,
+      topic: topic,
+      message: { value: JSON.stringify(Object.assign({}, msg, { timestamp: timestamp })) }
+    };
+  };
+
+  var send = function () {
+    var _ref2 = _asyncToGenerator(regeneratorRuntime.mark(function _callee2(topic, payload) {
+      var data, sendParams;
+      return regeneratorRuntime.wrap(function _callee2$(_context2) {
+        while (1) {
+          switch (_context2.prev = _context2.next) {
+            case 0:
+              logger.info('Message ', JSON.stringify(payload, null, 2), ' arrived on topic: ', topic);
+              data = producerPayload(payload, topic);
+
+              logger.info('*****', data);
+              sendParams = {
+                retries: {
+                  attempts: env.KAFKA_SEND_ATTEMPTS,
+                  delay: {
+                    min: env.KAFKA_SEND_DELAY_MIN,
+                    max: env.KAFKA_SEND_DELAY_MAX
+                  }
+                }
+              };
+              _context2.prev = 4;
+              _context2.next = 7;
+              return kafkaClient.producer.send(data, sendParams);
+
+            case 7:
+              logger.info('\n        *****PRODUCE********\n        topic:- ' + topic + '\n        ********************\n        payload:- ' + JSON.stringify(payload) + '\n        ********************\n      ');
+              _context2.next = 13;
+              break;
+
+            case 10:
+              _context2.prev = 10;
+              _context2.t0 = _context2['catch'](4);
+
+              logger.error('Error while sending payload:', JSON.stringify(payload, null, 2), 'topic :', topic, 'Error :', _context2.t0);
+
+            case 13:
+            case 'end':
+              return _context2.stop();
+          }
+        }
+      }, _callee2, undefined, [[4, 10]]);
+    }));
+
+    return function send(_x3, _x4) {
+      return _ref2.apply(this, arguments);
+    };
+  }();
+
   return {
     send: send,
-    receive: receive
+    receive: receive,
+    kafkaClient: kafkaClient,
+    initializeConsumer: initializeConsumer
   };
@@ -47,0 +141,0 @@ };

lib/task.js

@@ -7,29 +7,32 @@ 'use strict';
 
+var _constants = require('./constants');
+
+var _constants2 = _interopRequireDefault(_constants);
+
+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 Task = function Task(registry, runner) {
-  var topic = null;
-  var subscribeCallback = function subscribeCallback() {};
+var Task = function Task(registry, runner, logger) {
+  var topic = void 0;
+  var subscribeCallback = _constants2.default.NOOP;
 
   var subscribe = function subscribe(payload) {
     // publish message on topic without delay
-    console.log(payload);
+    logger.info(payload);
     return subscribeCallback(payload);
   };
 
-  var define = function define(taskName, callback) {
-    topic = taskName;
-    subscribeCallback = callback;
-    registry.addNewTask({
+  var define = function define(topicName, callBack) {
+    topic = topicName;
+    subscribeCallback = callBack;
+    var task = {
       topic: topic,
-      subscribe: subscribe
-    });
+      subscribe: subscribeCallback
+    };
+    registry.addNewTask(task, runner);
   };
 
   var publish = function () {
-    var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee() {
-      for (var _len = arguments.length, args = Array(_len), _key = 0; _key < _len; _key++) {
-        args[_key] = arguments[_key];
-      }
-
+    var _ref = _asyncToGenerator(regeneratorRuntime.mark(function _callee(payload) {
       return regeneratorRuntime.wrap(function _callee$(_context) {
@@ -39,8 +42,9 @@ while (1) {
             case 0:
-              // check with registry for valid topic
-              // publish message on topic
-              runner.send.apply(runner, args);
-              console.log(args, topic);
+              _context.next = 2;
+              return runner.send(topic, payload);
 
             case 2:
+              logger.info(topic, payload);
+
+            case 3:
             case 'end':
@@ -53,3 +57,3 @@ return _context.stop();
 
-    return function publish() {
+    return function publish(_x) {
       return _ref.apply(this, arguments);
@@ -65,2 +69,3 @@ };
 };
+
 exports.default = Task;

package.json

 {
   "name": "steveo",
-  "version": "1.0.4",
+  "version": "1.0.5",
   "description": "A Task Manager Library",
@@ -23,5 +23,16 @@ "main": "lib/index.js",
   ],
+  "nyc": {
+    "exclude": [
+      "lib",
+      "test",
+      "types",
+      "src/constants"
+    ]
+  },
   "dependencies": {
     "babel-polyfill": "^6.23.0",
-    "babel-runtime": "^6.23.0"
+    "babel-runtime": "^6.23.0",
+    "moment": "^2.18.1",
+    "no-kafka": "^3.1.1",
+    "uuid": "^3.0.1"
   },
@@ -28,0 +39,0 @@ "devDependencies": {

README.md

@@ -1,3 +0,11 @@
-## Steveo
+# Lunartick
+[![npm version](https://badge.fury.io/js/steveo.svg)](https://badge.fury.io/js/steveo)
+[![Build Status](https://travis-ci.org/ordermentum/steveo.svg?branch=master)](https://travis-ci.org/ordermentum/steveo)
+[![dependencies Status](https://david-dm.org/ordermentum/steveo/status.svg)](https://david-dm.org/ordermentum/steveo)
+[![devDependencies Status](https://david-dm.org/ordermentum/steveo/dev-status.svg)](https://david-dm.org/ordermentum/steveo?type=dev)
+[![npm](https://img.shields.io/npm/l/steveo.svg)](https://www.npmjs.com/package/steveo)
+[![npm](https://img.shields.io/npm/dt/steveo.svg)](https://www.npmjs.com/package/steveo)
 
+
+
 Status: WIP
@@ -9,15 +17,15 @@
 
-    +-----------+                  +-----------+
-    |           |                  |           |
-    |   TASK    |                  | REGISTRY  |
-    |           |                  |           |
-    |           |                  |           |
-    +-----------+                  +-----------+
+              +-----------+                  +-----------+
+              |           |                  |           |
+              |   TASK    |                  | REGISTRY  |
+              |           |                  |           |
+              |           |                  |           |
+              +-----------+                  +-----------+
 
-                    +-----------+
-                    |           |
-                    |   RUNNER  |
-        SEND  <-----|           |<----- RECEIVE
-                    |           |
-                    +-----------+
+                              +-----------+
+                              |           |
+                              |   RUNNER  |
+                  SEND  <-----|           |<----- RECEIVE
+                              |           |
+                              +-----------+
 
@@ -24,0 +32,0 @@ ### Task

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

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

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

23859

increased by218.37%

Number of package files

10

increased by66.67%

Lines of code

251

increased by100.8%

Number of lines in readme file

48

increased by20%

Worsened metrics

Dependency count

5

increased by150%

Number of medium supply chain risk alerts

1

increased byInfinity%

Dependency changes

• + Addedmoment@^2.18.1

• + Addedno-kafka@^3.1.1

• + Addeduuid@^3.0.1

• + Added@types/bluebird@3.5.0(transitive)

• + Added@types/lodash@4.17.13(transitive)

• + Addedbin-protocol@3.1.1(transitive)

• + Addedbluebird@3.7.2(transitive)

• + Addedbuffer-crc32@0.2.13(transitive)

• + Addedconnection-parse@0.0.7(transitive)

• + Addedhashring@3.2.0(transitive)

• + Addedlong@4.0.0(transitive)

• + Addedmoment@2.30.1(transitive)

• + Addedmurmur-hash-js@1.0.0(transitive)

• + Addednice-simple-logger@1.0.1(transitive)

• + Addedno-kafka@3.4.4(transitive)

• + Addedprotocol-buffers-schema@3.6.0(transitive)

• + Addedsimple-lru-cache@0.0.2(transitive)

• + Addeduuid@3.4.0(transitive)

• + Addedwrr-pool@1.1.4(transitive)