@lskjs/worker/lib/Worker.js

"use strict";

var _interopRequireDefault = require("@babel/runtime/helpers/interopRequireDefault");
Object.defineProperty(exports, "__esModule", {
  value: true
});
exports["default"] = exports.Worker = void 0;
var _regenerator = _interopRequireDefault(require("@babel/runtime/regenerator"));
var _toConsumableArray2 = _interopRequireDefault(require("@babel/runtime/helpers/toConsumableArray"));
var _defineProperty2 = _interopRequireDefault(require("@babel/runtime/helpers/defineProperty"));
var _asyncToGenerator2 = _interopRequireDefault(require("@babel/runtime/helpers/asyncToGenerator"));
var _classCallCheck2 = _interopRequireDefault(require("@babel/runtime/helpers/classCallCheck"));
var _createClass2 = _interopRequireDefault(require("@babel/runtime/helpers/createClass"));
var _get2 = _interopRequireDefault(require("@babel/runtime/helpers/get"));
var _inherits2 = _interopRequireDefault(require("@babel/runtime/helpers/inherits"));
var _possibleConstructorReturn2 = _interopRequireDefault(require("@babel/runtime/helpers/possibleConstructorReturn"));
var _getPrototypeOf2 = _interopRequireDefault(require("@babel/runtime/helpers/getPrototypeOf"));
var _err = _interopRequireDefault(require("@lskjs/err"));
var _module = _interopRequireDefault(require("@lskjs/module"));
var _stats = _interopRequireDefault(require("@lskjs/stats"));
var _env = require("@lskjs/utils/env");
var _tryJSONparse = _interopRequireDefault(require("@lskjs/utils/tryJSONparse"));
var _bluebird = _interopRequireDefault(require("bluebird"));
var _get3 = _interopRequireDefault(require("lodash/get"));
var _pick = _interopRequireDefault(require("lodash/pick"));
var _createTelegramMessage = require("./utils/createTelegramMessage");
function ownKeys(object, enumerableOnly) { var keys = Object.keys(object); if (Object.getOwnPropertySymbols) { var symbols = Object.getOwnPropertySymbols(object); enumerableOnly && (symbols = symbols.filter(function (sym) { return Object.getOwnPropertyDescriptor(object, sym).enumerable; })), keys.push.apply(keys, symbols); } return keys; }
function _objectSpread(target) { for (var i = 1; i < arguments.length; i++) { var source = null != arguments[i] ? arguments[i] : {}; i % 2 ? ownKeys(Object(source), !0).forEach(function (key) { (0, _defineProperty2["default"])(target, key, source[key]); }) : Object.getOwnPropertyDescriptors ? Object.defineProperties(target, Object.getOwnPropertyDescriptors(source)) : ownKeys(Object(source)).forEach(function (key) { Object.defineProperty(target, key, Object.getOwnPropertyDescriptor(source, key)); }); } return target; }
function _createSuper(Derived) { var hasNativeReflectConstruct = _isNativeReflectConstruct(); return function _createSuperInternal() { var Super = (0, _getPrototypeOf2["default"])(Derived), result; if (hasNativeReflectConstruct) { var NewTarget = (0, _getPrototypeOf2["default"])(this).constructor; result = Reflect.construct(Super, arguments, NewTarget); } else { result = Super.apply(this, arguments); } return (0, _possibleConstructorReturn2["default"])(this, result); }; }
function _isNativeReflectConstruct() { if (typeof Reflect === "undefined" || !Reflect.construct) return false; if (Reflect.construct.sham) return false; if (typeof Proxy === "function") return true; try { Boolean.prototype.valueOf.call(Reflect.construct(Boolean, [], function () {})); return true; } catch (e) { return false; } }
var Worker = /*#__PURE__*/function (_Module) {
  (0, _inherits2["default"])(Worker, _Module);
  var _super = _createSuper(Worker);
  function Worker() {
    var _this;
    (0, _classCallCheck2["default"])(this, Worker);
    for (var _len = arguments.length, args = new Array(_len), _key = 0; _key < _len; _key++) {
      args[_key] = arguments[_key];
    }
    _this = _super.call.apply(_super, [this].concat(args));
    _this.__worker = true;
    return _this;
  }
  (0, _createClass2["default"])(Worker, [{
    key: "showErrorInfo",
    value: function showErrorInfo() {
      return false; // // this.debug
    }
  }, {
    key: "getConfig",
    value: function () {
      var _getConfig = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee() {
        return _regenerator["default"].wrap(function _callee$(_context) {
          while (1) switch (_context.prev = _context.next) {
            case 0:
              _context.t0 = _objectSpread;
              _context.t1 = {
                autoconnect: true
              };
              _context.next = 4;
              return (0, _get2["default"])((0, _getPrototypeOf2["default"])(Worker.prototype), "getConfig", this).call(this);
            case 4:
              _context.t2 = _context.sent;
              return _context.abrupt("return", (0, _context.t0)(_context.t1, _context.t2));
            case 6:
            case "end":
              return _context.stop();
          }
        }, _callee, this);
      }));
      function getConfig() {
        return _getConfig.apply(this, arguments);
      }
      return getConfig;
    }()
  }, {
    key: "getJobConfig",
    value: function () {
      var _getJobConfig = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee2() {
        var _this$config, _this$config2, _this$config2$log;
        var mConfig, res;
        return _regenerator["default"].wrap(function _callee2$(_context2) {
          while (1) switch (_context2.prev = _context2.next) {
            case 0:
              _context2.next = 2;
              return this.getModuleConfig('job');
            case 2:
              mConfig = _context2.sent;
              res = _objectSpread(_objectSpread(_objectSpread({}, mConfig || {}), this.config || {}), {}, {
                log: _objectSpread(_objectSpread(_objectSpread({}, (mConfig === null || mConfig === void 0 ? void 0 : mConfig.log) || {}), ((_this$config = this.config) === null || _this$config === void 0 ? void 0 : _this$config.log) || {}), {}, {
                  ns: [(_this$config2 = this.config) === null || _this$config2 === void 0 ? void 0 : (_this$config2$log = _this$config2.log) === null || _this$config2$log === void 0 ? void 0 : _this$config2$log.ns, 'job'].filter(Boolean).join('.')
                })
              });
              return _context2.abrupt("return", res);
            case 5:
            case "end":
              return _context2.stop();
          }
        }, _callee2, this);
      }));
      function getJobConfig() {
        return _getJobConfig.apply(this, arguments);
      }
      return getJobConfig;
    }()
  }, {
    key: "parse",
    value: function () {
      var _parse = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee3() {
        return _regenerator["default"].wrap(function _callee3$(_context3) {
          while (1) switch (_context3.prev = _context3.next) {
            case 0:
              throw new _err["default"]('NOT_IMPLEMENTED', 'not implemented worker.parse()');
            case 1:
            case "end":
              return _context3.stop();
          }
        }, _callee3);
      }));
      function parse() {
        return _parse.apply(this, arguments);
      }
      return parse;
    }()
  }, {
    key: "onTelegramError",
    value: function () {
      var _onTelegramError = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee4(_ref) {
        var err, job, rlog;
        return _regenerator["default"].wrap(function _callee4$(_context4) {
          while (1) switch (_context4.prev = _context4.next) {
            case 0:
              err = _ref.err, job = _ref.job;
              if (!this.app.hasModule('rlog')) {
                _context4.next = 6;
                break;
              }
              _context4.next = 4;
              return this.app.module('rlog');
            case 4:
              rlog = _context4.sent;
              rlog.error((0, _createTelegramMessage.createTelegramMessage)({
                name: this.name,
                err: err,
                job: job
              }));
            case 6:
            case "end":
              return _context4.stop();
          }
        }, _callee4, this);
      }));
      function onTelegramError(_x) {
        return _onTelegramError.apply(this, arguments);
      }
      return onTelegramError;
    }()
  }, {
    key: "onConsume",
    value: function () {
      var _onConsume = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee5(msg) {
        var params, job;
        return _regenerator["default"].wrap(function _callee5$(_context5) {
          while (1) switch (_context5.prev = _context5.next) {
            case 0:
              if (this.showErrorInfo()) this.log.trace('[onConsume] start', msg);
              this.stats.print({
                log: this.log.info.bind(this.log),
                successKey: 'event.success',
                unsuccessKey: 'event.error'
              });
              if (!(msg === null)) {
                _context5.next = 4;
                break;
              }
              return _context5.abrupt("return");
            case 4:
              _context5.prev = 4;
              params = this.getMsgData(msg, 'content');
              _context5.next = 14;
              break;
            case 8:
              _context5.prev = 8;
              _context5.t0 = _context5["catch"](4);
              _context5.next = 12;
              return this.client.nack(msg, {
                requeue: false
              });
            case 12:
              if (this.showErrorInfo()) this.log.error('[onConsume] !content');
              return _context5.abrupt("return");
            case 14:
              _context5.next = 16;
              return this.createJob({
                msg: msg,
                params: params
              });
            case 16:
              job = _context5.sent;
              _context5.prev = 17;
              if (this.showErrorInfo()) this.log.trace('[onConsume] job.start()');
              _context5.next = 21;
              return job.start();
            case 21:
              if (this.showErrorInfo()) this.log.trace('[onConsume] job.start() finished');
              if (job.status) {
                _context5.next = 25;
                break;
              }
              _context5.next = 25;
              return job.ackSuccess();
            case 25:
              _context5.next = 38;
              break;
            case 27:
              _context5.prev = 27;
              _context5.t1 = _context5["catch"](17);
              _context5.prev = 29;
              if (this.showErrorInfo()) this.log.trace('[onConsume] onConsumeError');
              _context5.next = 33;
              return this.onConsumeError(_context5.t1, {
                job: job,
                msg: msg
              });
            case 33:
              _context5.next = 38;
              break;
            case 35:
              _context5.prev = 35;
              _context5.t2 = _context5["catch"](29);
              if (this.showErrorInfo()) this.log.error('[onConsume] onConsumeError err', _context5.t2);
            case 38:
              if (this.showErrorInfo()) this.log.trace('[onConsume] finish');
            case 39:
            case "end":
              return _context5.stop();
          }
        }, _callee5, this, [[4, 8], [17, 27], [29, 35]]);
      }));
      function onConsume(_x2) {
        return _onConsume.apply(this, arguments);
      }
      return onConsume;
    }()
  }, {
    key: "onConsumeError",
    value: function () {
      var _onConsumeError = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee7(error, _ref2) {
        var _this2 = this;
        var job, err, errorParams, _ref3, _ref3$delay, delay, errInfo, isApm, isNack, isTelegram, log, timeout, doTimeout, _this$log, code, message, apm, fromQueue, queue;
        return _regenerator["default"].wrap(function _callee7$(_context7) {
          while (1) switch (_context7.prev = _context7.next) {
            case 0:
              job = _ref2.job;
              if (this.showErrorInfo()) {
                this.log.error('-----------\n', error, '\n-----------'); // eslint-disable-line no-console
                this.log.error('-----------\n', error.stack, '\n-----------'); // eslint-disable-line no-console
              }
              _context7.prev = 2;
              errorParams = (0, _pick["default"])(error, 'nack', 'es', 'telegram', 'log');
              err = new _err["default"](error, errorParams);
              _ref3 = this.config.options || {}, _ref3$delay = _ref3.delay, delay = _ref3$delay === void 0 ? _env.isDev ? 10000 : 0 : _ref3$delay;
              if (!delay) {
                _context7.next = 10;
                break;
              }
              this.log.warn('[delay] 10000');
              _context7.next = 10;
              return _bluebird["default"].delay(10000);
            case 10:
              _context7.next = 15;
              break;
            case 12:
              _context7.prev = 12;
              _context7.t0 = _context7["catch"](2);
              this.log.error('error while onError', _context7.t0);
            case 15:
              if (job) {
                _context7.next = 17;
                break;
              }
              return _context7.abrupt("return");
            case 17:
              errInfo = this.app && this.app.getErrorInfo ? this.app.getErrorInfo(err) : {};
              if (this.showErrorInfo()) this.log.trace('onConsumeError errInfo', errInfo);
              isApm = (0, _get3["default"])(errInfo, 'apm', true);
              isNack = (0, _get3["default"])(errInfo, 'nack', true);
              isTelegram = (0, _get3["default"])(errInfo, 'telegram', true);
              log = (0, _get3["default"])(errInfo, 'log', 'error');
              timeout = errInfo.timeout || (0, _tryJSONparse["default"])(process.env.AMQP_ERROR_TIMEOUT) || (0, _tryJSONparse["default"])(process.env.ERROR_TIMEOUT);
              doTimeout = /*#__PURE__*/function () {
                var _ref4 = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee6() {
                  return _regenerator["default"].wrap(function _callee6$(_context6) {
                    while (1) switch (_context6.prev = _context6.next) {
                      case 0:
                        if (!timeout) {
                          _context6.next = 4;
                          break;
                        }
                        _this2.log.trace('[err.timeout]', timeout, '[delay]');
                        _context6.next = 4;
                        return _bluebird["default"].delay(timeout);
                      case 4:
                      case "end":
                        return _context6.stop();
                    }
                  }, _callee6);
                }));
                return function doTimeout() {
                  return _ref4.apply(this, arguments);
                };
              }();
              if (log && this.log[log]) {
                code = _err["default"].getCode(err);
                //= Err.getMessage(err);
                if (code === message) message = null;
                (_this$log = this.log)[log].apply(_this$log, (0, _toConsumableArray2["default"])([code, message].filter(Boolean)));
              }
              _context7.next = 28;
              return doTimeout();
            case 28:
              if (isTelegram) {
                this.onTelegramError({
                  err: err,
                  job: job
                });
              }
              if (!(isApm && this.app.hasModule('apm'))) {
                _context7.next = 34;
                break;
              }
              _context7.next = 32;
              return this.app.module('apm');
            case 32:
              apm = _context7.sent;
              try {
                apm.captureError(err);
              } catch (apmErr) {
                this.log.error('apm.captureError', apmErr);
              }
            case 34:
              if (isNack) {
                _context7.next = 39;
                break;
              }
              if (this.showErrorInfo()) console.error('err4', err); // eslint-disable-line no-console
              _context7.next = 38;
              return job.ackError(err);
            case 38:
              return _context7.abrupt("return");
            case 39:
              if (!(errInfo.redelivered && job.isTooMuchRedelivered && job.isTooMuchRedelivered())) {
                _context7.next = 49;
                break;
              }
              fromQueue = this.getQueue();
              queue = "".concat(fromQueue, "_redelivered");
              if (!this.sendToRedelivered) {
                _context7.next = 45;
                break;
              }
              _context7.next = 45;
              return this.sendToRedelivered(job, queue, {
                err: err,
                fromQueue: fromQueue
              });
            case 45:
              if (this.showErrorInfo()) console.error('err2', err); // eslint-disable-line no-console
              _context7.next = 48;
              return job.ackError(err);
            case 48:
              return _context7.abrupt("return");
            case 49:
              if (this.showErrorInfo()) console.error('err3', err); // eslint-disable-line no-console
              _context7.next = 52;
              return job.nackError(err);
            case 52:
            case "end":
              return _context7.stop();
          }
        }, _callee7, this, [[2, 12]]);
      }));
      function onConsumeError(_x3, _x4) {
        return _onConsumeError.apply(this, arguments);
      }
      return onConsumeError;
    }()
  }, {
    key: "initStats",
    value: function () {
      var _initStats = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee8() {
        var statsProps;
        return _regenerator["default"].wrap(function _callee8$(_context8) {
          while (1) switch (_context8.prev = _context8.next) {
            case 0:
              statsProps = {};
              statsProps.debug = this.log.trace.bind(this.log);
              _context8.next = 4;
              return _stats["default"].create(statsProps);
            case 4:
              this.stats = _context8.sent;
              if (this.stats) this.stats.startTimer();
            case 6:
            case "end":
              return _context8.stop();
          }
        }, _callee8, this);
      }));
      function initStats() {
        return _initStats.apply(this, arguments);
      }
      return initStats;
    }()
  }, {
    key: "getQueue",
    value: function getQueue() {
      return this.config.queue || this.queue || this.config.topic || this.topic;
    }
  }, {
    key: "getMsgData",
    value: function getMsgData(msg) {
      try {
        return JSON.parse(msg.toString());
      } catch (err) {
        var str = msg.toString();
        if (_env.isDev) {
          this.log.error('[ignore] cantParseJSON', str);
        } else {
          this.log.error('[ignore] cantParseJSON ');
          this.onTelegramError({
            err: 'cantParseJSON',
            job: {
              params: str
            }
          });
        }
        throw new _err["default"]('catntParseJSON', {
          str: str
        });
      }
    }
  }, {
    key: "getConsumerOptions",
    value: function getConsumerOptions() {
      return {};
    }
  }, {
    key: "connect",
    value: function () {
      var _connect = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee9() {
        var queue, onConsume;
        return _regenerator["default"].wrap(function _callee9$(_context9) {
          while (1) switch (_context9.prev = _context9.next) {
            case 0:
              _context9.next = 2;
              return this.app.module(this.__broker);
            case 2:
              this.client = _context9.sent;
              queue = this.getQueue();
              if (queue) {
                _context9.next = 7;
                break;
              }
              this.log.warn('!queue || !topic');
              return _context9.abrupt("return", null);
            case 7:
              onConsume = this.onConsume.bind(this);
              _context9.next = 10;
              return this.client.consume(queue, onConsume, this.getConsumerOptions());
            case 10:
              this.consumer = _context9.sent;
              return _context9.abrupt("return", this.consumer);
            case 12:
            case "end":
              return _context9.stop();
          }
        }, _callee9, this);
      }));
      function connect() {
        return _connect.apply(this, arguments);
      }
      return connect;
    }()
  }, {
    key: "run",
    value: function () {
      var _run = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee10() {
        var queue;
        return _regenerator["default"].wrap(function _callee10$(_context10) {
          while (1) switch (_context10.prev = _context10.next) {
            case 0:
              _context10.next = 2;
              return (0, _get2["default"])((0, _getPrototypeOf2["default"])(Worker.prototype), "run", this).call(this);
            case 2:
              queue = this.getQueue();
              if (queue) {
                _context10.next = 6;
                break;
              }
              this.log.warn('!queue || !topic');
              return _context10.abrupt("return");
            case 6:
              if (this.Job) {
                _context10.next = 9;
                break;
              }
              this.log.warn('!Job', this.Job);
              return _context10.abrupt("return");
            case 9:
              if (!this.config.autoconnect) {
                _context10.next = 15;
                break;
              }
              this.log.debug('autoconnect');
              _context10.next = 13;
              return this.initStats();
            case 13:
              _context10.next = 15;
              return this.connect();
            case 15:
            case "end":
              return _context10.stop();
          }
        }, _callee10, this);
      }));
      function run() {
        return _run.apply(this, arguments);
      }
      return run;
    }()
  }, {
    key: "createJob",
    value: function () {
      var _createJob = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee11() {
        var props,
          mergedProps,
          _args11 = arguments;
        return _regenerator["default"].wrap(function _callee11$(_context11) {
          while (1) switch (_context11.prev = _context11.next) {
            case 0:
              props = _args11.length > 0 && _args11[0] !== undefined ? _args11[0] : {};
              _context11.t0 = _objectSpread;
              _context11.t1 = new Date();
              _context11.t2 = this;
              _context11.t3 = this;
              _context11.t4 = this.app;
              _context11.next = 8;
              return this.getJobConfig();
            case 8:
              _context11.t5 = _context11.sent;
              _context11.t6 = {
                '__lifecycle.create': _context11.t1,
                worker: _context11.t2,
                __parent: _context11.t3,
                app: _context11.t4,
                config: _context11.t5
              };
              _context11.t7 = props;
              mergedProps = (0, _context11.t0)(_context11.t6, _context11.t7);
              if (this.client) {
                mergedProps.client = this.client;
                if (this.__broker) mergedProps[this.__broker] = this.client;
              }
              return _context11.abrupt("return", this.Job["new"](mergedProps));
            case 14:
            case "end":
              return _context11.stop();
          }
        }, _callee11, this);
      }));
      function createJob() {
        return _createJob.apply(this, arguments);
      }
      return createJob;
    }()
  }, {
    key: "process",
    value: function () {
      var _process = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee12(params) {
        var props,
          job,
          _args12 = arguments;
        return _regenerator["default"].wrap(function _callee12$(_context12) {
          while (1) switch (_context12.prev = _context12.next) {
            case 0:
              props = _args12.length > 1 && _args12[1] !== undefined ? _args12[1] : {};
              _context12.next = 3;
              return this.createJob(_objectSpread(_objectSpread({}, props), {}, {
                params: params
              }));
            case 3:
              job = _context12.sent;
              _context12.next = 6;
              return job.start();
            case 6:
              return _context12.abrupt("return", _objectSpread(_objectSpread({
                code: job.err ? _err["default"].getCode(job.err) : 0
              }, job.getQueueMeta()), {}, {
                startedAt: job.startedAt,
                finishedAt: job.finishedAt,
                runningTime: job.finishedAt ? job.finishedAt - job.startedAt : null,
                status: job.status,
                data: job.data,
                err: job.err
              }));
            case 7:
            case "end":
              return _context12.stop();
          }
        }, _callee12, this);
      }));
      function process(_x5) {
        return _process.apply(this, arguments);
      }
      return process;
    }()
  }, {
    key: "stop",
    value: function () {
      var _stop = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee13() {
        return _regenerator["default"].wrap(function _callee13$(_context13) {
          while (1) switch (_context13.prev = _context13.next) {
            case 0:
              _context13.next = 2;
              return (0, _get2["default"])((0, _getPrototypeOf2["default"])(Worker.prototype), "stop", this).call(this);
            case 2:
              return _context13.abrupt("return", this.client.stop());
            case 3:
            case "end":
              return _context13.stop();
          }
        }, _callee13, this);
      }));
      function stop() {
        return _stop.apply(this, arguments);
      }
      return stop;
    }()
  }]);
  return Worker;
}(_module["default"]);
exports.Worker = Worker;
Worker.__worker = true;
var _default = Worker;
exports["default"] = _default;
//# sourceMappingURL=Worker.js.map