strong-tie/inbound-calls

1'use strict'2 3const fastq = require('fastq')4const EE = require('node:events').EventEmitter5const inherits = require('node:util').inherits6const {7  AVV_ERR_EXPOSE_ALREADY_DEFINED,8  AVV_ERR_CALLBACK_NOT_FN,9  AVV_ERR_ROOT_PLG_BOOTED,10  AVV_ERR_READY_TIMEOUT,11  AVV_ERR_ATTRIBUTE_ALREADY_DEFINED12} = require('./lib/errors')13const {14  kAvvio,15  kIsOnCloseHandler16} = require('./lib/symbols')17const { TimeTree } = require('./lib/time-tree')18const { Plugin } = require('./lib/plugin')19const { debug } = require('./lib/debug')20const { validatePlugin } = require('./lib/validate-plugin')21const { isBundledOrTypescriptPlugin } = require('./lib/is-bundled-or-typescript-plugin')22const { isPromiseLike } = require('./lib/is-promise-like')23const { thenify } = require('./lib/thenify')24const { executeWithThenable } = require('./lib/execute-with-thenable')25 26function Boot (server, opts, done) {27  if (typeof server === 'function' && arguments.length === 1) {28    done = server29    opts = {}30    server = null31  }32 33  if (typeof opts === 'function') {34    done = opts35    opts = {}36  }37 38  opts = opts || {}39  opts.autostart = opts.autostart !== false40  opts.timeout = Number(opts.timeout) || 041  opts.expose = opts.expose || {}42 43  if (!new.target) {44    return new Boot(server, opts, done)45  }46 47  this._server = server || this48  this._opts = opts49 50  if (server) {51    this._expose()52  }53 54  /**55   * @type {Array<Plugin>}56   */57  this._current = []58 59  this._error = null60 61  this._lastUsed = null62 63  this.setMaxListeners(0)64 65  if (done) {66    this.once('start', done)67  }68 69  this.started = false70  this.booted = false71  this.pluginTree = new TimeTree()72 73  this._readyQ = fastq(this, callWithCbOrNextTick, 1)74  this._readyQ.pause()75  this._readyQ.drain = () => {76    this.emit('start')77    // nooping this, we want to emit start only once78    this._readyQ.drain = noop79  }80 81  this._closeQ = fastq(this, closeWithCbOrNextTick, 1)82  this._closeQ.pause()83  this._closeQ.drain = () => {84    this.emit('close')85    // nooping this, we want to emit close only once86    this._closeQ.drain = noop87  }88 89  this._doStart = null90 91  const instance = this92  this._root = new Plugin(fastq(this, this._loadPluginNextTick, 1), function root (server, opts, done) {93    instance._doStart = done94    opts.autostart && instance.start()95  }, opts, false, 0)96 97  this._trackPluginLoading(this._root)98 99  this._loadPlugin(this._root, (err) => {100    debug('root plugin ready')101    try {102      this.emit('preReady')103      this._root = null104    } catch (preReadyError) {105      err = err || this._error || preReadyError106    }107 108    if (err) {109      this._error = err110      if (this._readyQ.length() === 0) {111        throw err112      }113    } else {114      this.booted = true115    }116    this._readyQ.resume()117  })118}119 120inherits(Boot, EE)121 122Boot.prototype.start = function () {123  this.started = true124 125  // we need to wait any call to use() to happen126  process.nextTick(this._doStart)127  return this128}129 130// allows to override the instance of a server, given a plugin131Boot.prototype.override = function (server, func, opts) {132  return server133}134 135Boot.prototype[kAvvio] = true136 137// load a plugin138Boot.prototype.use = function (plugin, opts) {139  this._lastUsed = this._addPlugin(plugin, opts, false)140  return this141}142 143Boot.prototype._loadRegistered = function () {144  const plugin = this._current[0]145  const weNeedToStart = !this.started && !this.booted146 147  // if the root plugin is not loaded, let's resume that148  // so one can use after() before calling ready149  if (weNeedToStart) {150    process.nextTick(() => this._root.queue.resume())151  }152 153  if (!plugin) {154    return Promise.resolve()155  }156 157  return plugin.loadedSoFar()158}159 160Object.defineProperty(Boot.prototype, 'then', { get: thenify })161 162Boot.prototype._addPlugin = function (pluginFn, opts, isAfter) {163  if (isBundledOrTypescriptPlugin(pluginFn)) {164    pluginFn = pluginFn.default165  }166  validatePlugin(pluginFn)167  opts = opts || {}168 169  if (this.booted) {170    throw new AVV_ERR_ROOT_PLG_BOOTED()171  }172 173  // we always add plugins to load at the current element174  const current = this._current[0]175 176  let timeout = this._opts.timeout177 178  if (!current.loaded && current.timeout > 0) {179    const delta = Date.now() - current.startTime180    // We need to decrease it by 3ms to make sure the internal timeout181    // is triggered earlier than the parent182    timeout = current.timeout - (delta + 3)183  }184 185  const plugin = new Plugin(fastq(this, this._loadPluginNextTick, 1), pluginFn, opts, isAfter, timeout)186  this._trackPluginLoading(plugin)187 188  if (current.loaded) {189    throw new Error(plugin.name, current.name)190  }191 192  // we add the plugin to be loaded at the end of the current queue193  current.enqueue(plugin, (err) => { err && (this._error = err) })194 195  return plugin196}197 198Boot.prototype._expose = function _expose () {199  const instance = this200  const server = instance._server201  const {202    use: useKey = 'use',203    after: afterKey = 'after',204    ready: readyKey = 'ready',205    onClose: onCloseKey = 'onClose',206    close: closeKey = 'close'207  } = this._opts.expose208 209  if (server[useKey]) {210    throw new AVV_ERR_EXPOSE_ALREADY_DEFINED(useKey, 'use')211  }212  server[useKey] = function (fn, opts) {213    instance.use(fn, opts)214    return this215  }216 217  if (server[afterKey]) {218    throw new AVV_ERR_EXPOSE_ALREADY_DEFINED(afterKey, 'after')219  }220  server[afterKey] = function (func) {221    if (typeof func !== 'function') {222      return instance._loadRegistered()223    }224    instance.after(encapsulateThreeParam(func, this))225    return this226  }227 228  if (server[readyKey]) {229    throw new AVV_ERR_EXPOSE_ALREADY_DEFINED(readyKey, 'ready')230  }231  server[readyKey] = function (func) {232    if (func && typeof func !== 'function') {233      throw new AVV_ERR_CALLBACK_NOT_FN(readyKey, typeof func)234    }235    return instance.ready(func ? encapsulateThreeParam(func, this) : undefined)236  }237 238  if (server[onCloseKey]) {239    throw new AVV_ERR_EXPOSE_ALREADY_DEFINED(onCloseKey, 'onClose')240  }241  server[onCloseKey] = function (func) {242    if (typeof func !== 'function') {243      throw new AVV_ERR_CALLBACK_NOT_FN(onCloseKey, typeof func)244    }245    instance.onClose(encapsulateTwoParam(func, this))246    return this247  }248 249  if (server[closeKey]) {250    throw new AVV_ERR_EXPOSE_ALREADY_DEFINED(closeKey, 'close')251  }252  server[closeKey] = function (func) {253    if (func && typeof func !== 'function') {254      throw new AVV_ERR_CALLBACK_NOT_FN(closeKey, typeof func)255    }256 257    if (func) {258      instance.close(encapsulateThreeParam(func, this))259      return this260    }261 262    // this is a Promise263    return instance.close()264  }265 266  if (server.then) {267    throw new AVV_ERR_ATTRIBUTE_ALREADY_DEFINED('then')268  }269  Object.defineProperty(server, 'then', { get: thenify.bind(instance) })270 271  server[kAvvio] = true272}273 274Boot.prototype.after = function (func) {275  if (!func) {276    return this._loadRegistered()277  }278 279  this._addPlugin(_after.bind(this), {}, true)280 281  function _after (s, opts, done) {282    callWithCbOrNextTick.call(this, func, done)283  }284 285  return this286}287 288Boot.prototype.onClose = function (func) {289  // this is used to distinguish between onClose and close handlers290  // because they share the same queue but must be called with different signatures291 292  if (typeof func !== 'function') {293    throw new AVV_ERR_CALLBACK_NOT_FN('onClose', typeof func)294  }295 296  func[kIsOnCloseHandler] = true297  this._closeQ.unshift(func, (err) => { err && (this._error = err) })298 299  return this300}301 302Boot.prototype.close = function (func) {303  let promise304 305  if (func) {306    if (typeof func !== 'function') {307      throw new AVV_ERR_CALLBACK_NOT_FN('close', typeof func)308    }309  } else {310    promise = new Promise(function (resolve, reject) {311      func = function (err) {312        if (err) {313          return reject(err)314        }315        resolve()316      }317    })318  }319 320  this.ready(() => {321    this._error = null322    this._closeQ.push(func)323    process.nextTick(this._closeQ.resume.bind(this._closeQ))324  })325 326  return promise327}328 329Boot.prototype.ready = function (func) {330  if (func) {331    if (typeof func !== 'function') {332      throw new AVV_ERR_CALLBACK_NOT_FN('ready', typeof func)333    }334    this._readyQ.push(func)335    queueMicrotask(this.start.bind(this))336    return337  }338 339  return new Promise((resolve, reject) => {340    this._readyQ.push(readyPromiseCB)341    this.start()342 343    /**344     * The `encapsulateThreeParam` let callback function345     * bind to the right server instance.346     * In promises we need to track the last server347     * instance loaded, the first one in the _current queue.348     */349    const relativeContext = this._current[0].server350 351    function readyPromiseCB (err, context, done) {352      // the context is always binded to the root server353      if (err) {354        reject(err)355      } else {356        resolve(relativeContext)357      }358      process.nextTick(done)359    }360  })361}362 363/**364 * @param {Plugin} plugin365 * @returns {void}366 */367Boot.prototype._trackPluginLoading = function (plugin) {368  const parentName = this._current[0]?.name || null369  plugin.once('start', (serverName, funcName, time) => {370    const nodeId = this.pluginTree.start(parentName || null, funcName, time)371    plugin.once('loaded', (serverName, funcName, time) => {372      this.pluginTree.stop(nodeId, time)373    })374  })375}376 377Boot.prototype.prettyPrint = function () {378  return this.pluginTree.prettyPrint()379}380 381Boot.prototype.toJSON = function () {382  return this.pluginTree.toJSON()383}384 385/**386 * @callback LoadPluginCallback387 * @param {Error} [err]388 */389 390/**391 * Load a plugin392 *393 * @param {Plugin} plugin394 * @param {LoadPluginCallback} callback395 */396Boot.prototype._loadPlugin = function (plugin, callback) {397  const instance = this398  if (isPromiseLike(plugin.func)) {399    plugin.func.then((fn) => {400      if (typeof fn.default === 'function') {401        fn = fn.default402      }403      plugin.func = fn404      this._loadPlugin(plugin, callback)405    }, callback)406    return407  }408 409  const last = instance._current[0]410 411  // place the plugin at the top of _current412  instance._current.unshift(plugin)413 414  if (instance._error && !plugin.isAfter) {415    debug('skipping loading of plugin as instance errored and it is not an after', plugin.name)416    process.nextTick(execCallback)417    return418  }419 420  let server = (last && last.server) || instance._server421 422  if (!plugin.isAfter) {423    // Skip override for after424    try {425      server = instance.override(server, plugin.func, plugin.options)426    } catch (overrideErr) {427      debug('override errored', plugin.name)428      return execCallback(overrideErr)429    }430  }431 432  plugin.exec(server, execCallback)433 434  function execCallback (err) {435    plugin.finish(err, (err) => {436      instance._current.shift()437      callback(err)438    })439  }440}441 442/**443* Delays plugin loading until the next tick to ensure any bound `_after` callbacks have a chance444* to run prior to executing the next plugin445*/446Boot.prototype._loadPluginNextTick = function (plugin, callback) {447  process.nextTick(this._loadPlugin.bind(this), plugin, callback)448}449 450function noop () { }451 452function callWithCbOrNextTick (func, cb) {453  const context = this._server454  const err = this._error455 456  // with this the error will appear just in the next after/ready callback457  this._error = null458  if (func.length === 0) {459    this._error = err460    executeWithThenable(func, [], cb)461  } else if (func.length === 1) {462    executeWithThenable(func, [err], cb)463  } else {464    if (this._opts.timeout === 0) {465      const wrapCb = (err) => {466        this._error = err467        cb(this._error)468      }469 470      if (func.length === 2) {471        func(err, wrapCb)472      } else {473        func(err, context, wrapCb)474      }475    } else {476      timeoutCall.call(this, func, err, context, cb)477    }478  }479}480 481function timeoutCall (func, rootErr, context, cb) {482  const name = func.unwrappedName ?? func.name483  debug('setting up ready timeout', name, this._opts.timeout)484  let timer = setTimeout(() => {485    debug('timed out', name)486    timer = null487    const toutErr = new AVV_ERR_READY_TIMEOUT(name)488    toutErr.fn = func489    this._error = toutErr490    cb(toutErr)491  }, this._opts.timeout)492 493  if (func.length === 2) {494    func(rootErr, timeoutCb.bind(this))495  } else {496    func(rootErr, context, timeoutCb.bind(this))497  }498 499  function timeoutCb (err) {500    if (timer) {501      clearTimeout(timer)502      this._error = err503      cb(this._error)504    } else {505      // timeout has been triggered506      // can not call cb twice507    }508  }509}510 511function closeWithCbOrNextTick (func, cb) {512  const context = this._server513  const isOnCloseHandler = func[kIsOnCloseHandler]514  if (func.length === 0 || func.length === 1) {515    let promise516    if (isOnCloseHandler) {517      promise = func(context)518    } else {519      promise = func(this._error)520    }521    if (promise && typeof promise.then === 'function') {522      debug('resolving close/onClose promise')523      promise.then(524        () => process.nextTick(cb),525        (e) => process.nextTick(cb, e))526    } else {527      process.nextTick(cb)528    }529  } else if (func.length === 2) {530    if (isOnCloseHandler) {531      func(context, cb)532    } else {533      func(this._error, cb)534    }535  } else {536    if (isOnCloseHandler) {537      func(context, cb)538    } else {539      func(this._error, context, cb)540    }541  }542}543 544function encapsulateTwoParam (func, that) {545  return _encapsulateTwoParam.bind(that)546  function _encapsulateTwoParam (context, cb) {547    let res548    if (func.length === 0) {549      res = func()550      if (res && res.then) {551        res.then(function () {552          process.nextTick(cb)553        }, cb)554      } else {555        process.nextTick(cb)556      }557    } else if (func.length === 1) {558      res = func(this)559 560      if (res && res.then) {561        res.then(function () {562          process.nextTick(cb)563        }, cb)564      } else {565        process.nextTick(cb)566      }567    } else {568      func(this, cb)569    }570  }571}572 573function encapsulateThreeParam (func, that) {574  const wrapped = _encapsulateThreeParam.bind(that)575  wrapped.unwrappedName = func.name576  return wrapped577  function _encapsulateThreeParam (err, cb) {578    let res579    if (!func) {580      process.nextTick(cb)581    } else if (func.length === 0) {582      res = func()583      if (res && res.then) {584        res.then(function () {585          process.nextTick(cb, err)586        }, cb)587      } else {588        process.nextTick(cb, err)589      }590    } else if (func.length === 1) {591      res = func(err)592      if (res && res.then) {593        res.then(function () {594          process.nextTick(cb)595        }, cb)596      } else {597        process.nextTick(cb)598      }599    } else if (func.length === 2) {600      func(err, cb)601    } else {602      func(err, this, cb)603    }604  }605}606 607module.exports = Boot608