@lskjs/worker/lib/WorkerJob.js

"use strict";

var _interopRequireDefault = require("@babel/runtime/helpers/interopRequireDefault");
Object.defineProperty(exports, "__esModule", {
  value: true
});
exports["default"] = exports.apmMock = exports.WorkerJob = void 0;
var _regenerator = _interopRequireDefault(require("@babel/runtime/regenerator"));
var _defineProperty2 = _interopRequireDefault(require("@babel/runtime/helpers/defineProperty"));
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 _asyncToGenerator2 = _interopRequireDefault(require("@babel/runtime/helpers/asyncToGenerator"));
var _env = require("@lskjs/env");
var _err = _interopRequireDefault(require("@lskjs/err"));
var _module = _interopRequireDefault(require("@lskjs/module"));
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; } }
// , { IModule, IModuleWithApp }

var apmMock = {
  startTransaction: function startTransaction() {
    return {
      end: function end() {
        return null;
      },
      startSpan: function startSpan() {
        return {
          end: function end() {
            return null;
          }
        };
      },
      addLabels: function addLabels() {
        return null;
      },
      setLabel: function setLabel() {
        return null;
      }
    };
  },
  captureError: function captureError() {
    return (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:
            return _context.abrupt("return", null);
          case 1:
          case "end":
            return _context.stop();
        }
      }, _callee);
    }))();
  },
  setCustomContext: function setCustomContext() {
    return null;
  },
  setLabel: function setLabel() {
    return null;
  },
  addLabels: function addLabels() {
    return null;
  }
};

// export interface IRabbitWorker extends IModule {
//   name?: string;
//   stats?: any;
// }
// export interface IWorkerJob extends IModule {
//   worker?: IRabbitWorker;
//   stats?: any;
//   startedAt: Date,
//   tx: any,
// }

// implements IWorkerJob
exports.apmMock = apmMock;
var WorkerJob = /*#__PURE__*/function (_Module) {
  (0, _inherits2["default"])(WorkerJob, _Module);
  var _super = _createSuper(WorkerJob);
  function WorkerJob() {
    var _this;
    (0, _classCallCheck2["default"])(this, WorkerJob);
    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;
    _this.redeliveredCount = _env.isDev ? null : 10;
    return _this;
  }
  (0, _createClass2["default"])(WorkerJob, [{
    key: "_run",
    value: function () {
      var _run2 = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee2() {
        return _regenerator["default"].wrap(function _callee2$(_context2) {
          while (1) switch (_context2.prev = _context2.next) {
            case 0:
              if (this.__lifecycle.runStart) {
                _context2.next = 2;
                break;
              }
              throw new _err["default"]('MODULE_INVALID_LIVECYCLE_RUN', 'use module.__run() instead module.run()', {
                data: {
                  name: this.name
                }
              });
            case 2:
              _context2.next = 4;
              return this.run();
            case 4:
            case "end":
              return _context2.stop();
          }
        }, _callee2, this);
      }));
      function _run() {
        return _run2.apply(this, arguments);
      }
      return _run;
    }()
  }, {
    key: "init",
    value: function () {
      var _init = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee3() {
        var apm;
        return _regenerator["default"].wrap(function _callee3$(_context3) {
          while (1) switch (_context3.prev = _context3.next) {
            case 0:
              _context3.next = 2;
              return (0, _get2["default"])((0, _getPrototypeOf2["default"])(WorkerJob.prototype), "init", this).call(this);
            case 2:
              if (this.worker) this.stats = this.worker.stats;
              this.startedAt = new Date();
              if (!this.app.hasModule('apm')) {
                _context3.next = 8;
                break;
              }
              _context3.next = 7;
              return this.app.module('apm', {
                "throw": false
              });
            case 7:
              apm = _context3.sent;
            case 8:
              if (!apm) apm = apmMock;
              this.tx = apm.startTransaction(this.name, 'job');
            case 10:
            case "end":
              return _context3.stop();
          }
        }, _callee3, this);
      }));
      function init() {
        return _init.apply(this, arguments);
      }
      return init;
    }()
  }, {
    key: "getQueueMeta",
    value: function getQueueMeta() {
      return {
        __meta: _objectSpread({
          __name: this.name,
          __date: Date.now()
        }, this.params)
      };
    }
  }, {
    key: "isTooMuchRedelivered",
    value: function isTooMuchRedelivered() {
      if (!this.msg) return false;
      if (!this.redeliveredCount) return false;
      return this.msg.fields.redelivered && this.msg.fields.deliveryTag > this.redeliveredCount;
    }
  }, {
    key: "success",
    value: function success(res) {
      return this.result(res);
    }
  }, {
    key: "error",
    value: function error(res) {
      return this.result(res);
    }
  }, {
    key: "result",
    value: function result(res) {
      this.data = res;
      return res;
    }
  }, {
    key: "onSuccess",
    value: function () {
      var _onSuccess = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee4() {
        return _regenerator["default"].wrap(function _callee4$(_context4) {
          while (1) switch (_context4.prev = _context4.next) {
            case 0:
              this.finishedAt = new Date();
              if (this.stats) this.stats.trigger({
                event: 'success',
                startedAt: this.startedAt
              });
              if (!this.tx) {
                _context4.next = 6;
                break;
              }
              this.tx.result = 'success';
              _context4.next = 6;
              return this.tx.end();
            case 6:
            case "end":
              return _context4.stop();
          }
        }, _callee4, this);
      }));
      function onSuccess() {
        return _onSuccess.apply(this, arguments);
      }
      return onSuccess;
    }()
  }, {
    key: "onAckError",
    value: function () {
      var _onAckError = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee5(err) {
        return _regenerator["default"].wrap(function _callee5$(_context5) {
          while (1) switch (_context5.prev = _context5.next) {
            case 0:
              this.err = err;
              this.finishedAt = new Date();
              if (this.stats) this.stats.trigger({
                event: 'skip',
                startedAt: this.startedAt
              });
              if (!this.tx) {
                _context5.next = 7;
                break;
              }
              this.tx.result = 'skip';
              _context5.next = 7;
              return this.tx.end();
            case 7:
            case "end":
              return _context5.stop();
          }
        }, _callee5, this);
      }));
      function onAckError(_x) {
        return _onAckError.apply(this, arguments);
      }
      return onAckError;
    }()
  }, {
    key: "onError",
    value: function () {
      var _onError = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee6(err) {
        return _regenerator["default"].wrap(function _callee6$(_context6) {
          while (1) switch (_context6.prev = _context6.next) {
            case 0:
              this.err = err;
              this.finishedAt = new Date();
              if (this.stats) this.stats.trigger({
                event: 'error',
                startedAt: this.startedAt
              });
              if (!this.tx) {
                _context6.next = 7;
                break;
              }
              // await this.apm.client.captureError(err);
              this.tx.result = 'error';
              _context6.next = 7;
              return this.tx.end();
            case 7:
            case "end":
              return _context6.stop();
          }
        }, _callee6, this);
      }));
      function onError(_x2) {
        return _onError.apply(this, arguments);
      }
      return onError;
    }()
  }, {
    key: "setStatus",
    value: function setStatus(status) {
      this.status = status;
      return status;
    }

    /**
     * Когда всё хорошо
     */
  }, {
    key: "ackSuccess",
    value: function () {
      var _ackSuccess = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee7() {
        return _regenerator["default"].wrap(function _callee7$(_context7) {
          while (1) switch (_context7.prev = _context7.next) {
            case 0:
              if (!this.status) {
                _context7.next = 3;
                break;
              }
              this.log.warn('ackSuccess ignore because has status');
              return _context7.abrupt("return", this.status);
            case 3:
              _context7.next = 5;
              return this.onSuccess();
            case 5:
              if (this.debug) this.log.trace('[ack]');
              if (!this.msg) {
                _context7.next = 9;
                break;
              }
              _context7.next = 9;
              return this.client.ack(this.msg);
            case 9:
              return _context7.abrupt("return", this.setStatus({
                status: 'success',
                action: 'ack'
              }));
            case 10:
            case "end":
              return _context7.stop();
          }
        }, _callee7, this);
      }));
      function ackSuccess() {
        return _ackSuccess.apply(this, arguments);
      }
      return ackSuccess;
    }()
    /**
     * Когда всё плохо
     */
  }, {
    key: "nackError",
    value: function () {
      var _nackError = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee8(err) {
        return _regenerator["default"].wrap(function _callee8$(_context8) {
          while (1) switch (_context8.prev = _context8.next) {
            case 0:
              if (!this.status) {
                _context8.next = 3;
                break;
              }
              this.log.warn('nackError ignore because has status');
              return _context8.abrupt("return", this.status);
            case 3:
              _context8.next = 5;
              return this.onError(err);
            case 5:
              // if (this.debug) this.log.error(`rabbit.nack [${Err.getCode(err)}]`, this.params);
              if (this.debug) this.log.error("[nack] ".concat(_err["default"].getCode(err)));
              // if (this.msg && this.worker.consumerTag === this.msg.fields.consumerTag) {
              if (!this.msg) {
                _context8.next = 9;
                break;
              }
              _context8.next = 9;
              return this.client.nack(this.msg, {
                requeue: true
              });
            case 9:
              return _context8.abrupt("return", this.setStatus({
                status: 'error',
                action: 'nack'
              }));
            case 10:
            case "end":
              return _context8.stop();
          }
        }, _callee8, this);
      }));
      function nackError(_x3) {
        return _nackError.apply(this, arguments);
      }
      return nackError;
    }()
    /**
     * Когда ошибка заебала, и больше это сообщение не нужно в очереди (например при множественно redelivered)
     */
  }, {
    key: "ackError",
    value: function () {
      var _ackError = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee9(err) {
        return _regenerator["default"].wrap(function _callee9$(_context9) {
          while (1) switch (_context9.prev = _context9.next) {
            case 0:
              if (!this.status) {
                _context9.next = 3;
                break;
              }
              this.log.warn('ackError ignore because has status');
              return _context9.abrupt("return", this.status);
            case 3:
              _context9.next = 5;
              return this.onAckError(err);
            case 5:
              if (this.debug) this.log.trace("[ackError] ".concat(_err["default"].getCode(err)));
              if (!this.msg) {
                _context9.next = 9;
                break;
              }
              _context9.next = 9;
              return this.client.nack(this.msg, {
                requeue: false
              });
            case 9:
              return _context9.abrupt("return", this.setStatus({
                status: 'error',
                action: 'nack',
                requeue: false
              }));
            case 10:
            case "end":
              return _context9.stop();
          }
        }, _callee9, this);
      }));
      function ackError(_x4) {
        return _ackError.apply(this, arguments);
      }
      return ackError;
    }()
    /**
     * Хз когда эта хрень нужна, просто добавил для консистентности
     * напишу throw, если вдруг кому-то реально понадобится -- пинганите меня
     * @isuvorov
     */
  }, {
    key: "nackSuccess",
    value: function () {
      var _nackSuccess = (0, _asyncToGenerator2["default"])( /*#__PURE__*/_regenerator["default"].mark(function _callee10() {
        return _regenerator["default"].wrap(function _callee10$(_context10) {
          while (1) switch (_context10.prev = _context10.next) {
            case 0:
              throw new _err["default"]('ХЗ КОМУ ЭТА ХРЕНЬ НУЖНА, ПИНГАНИТЕ МЕНЯ');
            case 1:
            case "end":
              return _context10.stop();
          }
        }, _callee10);
      }));
      function nackSuccess() {
        return _nackSuccess.apply(this, arguments);
      }
      return nackSuccess;
    }()
  }]);
  return WorkerJob;
}(_module["default"]);
exports.WorkerJob = WorkerJob;
WorkerJob.__worker = true;
var _default = WorkerJob;
exports["default"] = _default;
//# sourceMappingURL=WorkerJob.js.map