index.js

'use strict'

/**
 * Copyright 2016-present, Dustin Deus (deusdustin@gmail.com)
 * All rights reserved.
 *
 * This source code is licensed under the MIT-style license found in the
 * LICENSE file in the root directory of this source tree.
 *
 */

/**
 * Module Dependencies
 */

const NATS = require('nats')
const EventEmitter = require('events')
const Bloomrun = require('bloomrun')
const Errio = require('errio')
const Heavy = require('heavy')
const Pino = require('pino')
const TinySonic = require('tinysonic')
const SuperError = require('super-error')
const Joi = require('joi')
const Avvio = require('avvio')
const Stream = require('stream').Stream

const runExt = require('./extensionRunner').extRunner
const serverExtIterator = require('./extensionRunner').serverExtIterator
const clientExtIterator = require('./extensionRunner').clientExtIterator
const Errors = require('./errors')
const Symbols = require('./symbols')
const Util = require('./util')
const NatsTransport = require('./transport')
const DefaultExtensions = require('./extensions')
const DefaultEncoder = require('./encoder')
const DefaultDecoder = require('./decoder')
const ServerResponse = require('./serverResponse')
const ServerRequest = require('./serverRequest')
const ClientRequest = require('./clientRequest')
const ClientResponse = require('./clientResponse')
const ConfigScheme = require('./configScheme')
const Reply = require('./reply')
const Add = require('./add')
const ExtensionManager = require('./extensionManager')

const natsConnCodes = [
  NATS.CONN_ERR,
  NATS.SECURE_CONN_REQ,
  NATS.NON_SECURE_CONN_REQ,
  NATS.CLIENT_CERT_REQ
]

/**
 * @class Hemera
 */
class Hemera extends EventEmitter {
  /**
   * Creates an instance of Hemera
   *
   * @param {Nats} transport
   * @param {Config} params
   *
   * @memberOf Hemera
   */
  constructor(transport, params) {
    super()

    const config = Joi.validate(params || {}, ConfigScheme)
    if (config.error) {
      throw config.error
    }

    this._root = this
    this._isReady = false
    this._config = config.value
    this._router = Bloomrun(this._config.bloomrun)
    this._heavy = new Heavy(this._config.load.process)
    this._transport = new NatsTransport({
      transport
    })
    this._topics = new Map()

    // special variables for the new execution context
    this.context$ = {}
    this.meta$ = {}
    this.delegate$ = {}
    this.auth$ = {}
    this.trace$ = {}
    this.request$ = {
      type: 'request',
      id: ''
    }
    this.sid = 0

    // represent the 'Add' instance
    this.matchedAction = null
    this.request = null
    this.response = null

    this._topic = ''
    this._pattern = null
    // represent the 'act' handler
    this._execute = null
    this._cleanPattern = ''

    this._clientEncoder = DefaultEncoder.encode
    this._clientDecoder = DefaultDecoder.decode
    this._serverEncoder = DefaultEncoder.encode
    this._serverDecoder = DefaultDecoder.decode
    this._schemaCompiler = null
    this._responseSchemaCompiler = null
    this._idGenerator = Util.randomId

    // errio settings
    Errio.setDefaults(this._config.errio)

    // Register all default hemera errors
    this._registerErrors()

    // create load policy
    this._loadPolicy = this._heavy.policy(this._config.load.policy)

    // start tracking process stats
    this._heavy.start()

    // when a route for a pattern could not be found
    this._notFoundPattern = null

    this._onAddHandlers = []

    this._extensionManager = new ExtensionManager()
    this._extensionManager.add(
      'onClientPreRequest',
      DefaultExtensions.onClientPreRequest
    )
    this._extensionManager.add(
      'onClientPostRequest',
      DefaultExtensions.onClientPostRequest
    )
    this._extensionManager.add(
      'onServerPreRequest',
      DefaultExtensions.onServerPreRequest
    )
    this._extensionManager.add(
      'onServerPreResponse',
      DefaultExtensions.onServerPreResponse
    )

    this._configureLogger()

    this._avvio = Avvio(this, {
      autostart: false,
      expose: {
        use: 'register',
        close: 'shutdown',
        onClose: 'onShutdown',
        ready: 'bootstrap'
      },
      timeout: this._config.pluginTimeout
    })

    this[Symbols.childrenKey] = []
    this[Symbols.registeredPlugins] = []

    this._avvio.override = (hemera, plugin, opts) => {
      const pluginMeta = this.getPluginMeta(plugin)

      if (pluginMeta) {
        if (pluginMeta.name) {
          hemera[Symbols.registeredPlugins].push(pluginMeta.name)
        }
        hemera.checkPluginDependencies(plugin)
        hemera.checkPluginDecorators(plugin)
      }

      if (plugin[Symbol.for('plugin-scoped')] === false) {
        return hemera
      }

      const instance = Object.create(hemera)

      hemera[Symbols.childrenKey].push(instance)
      instance[Symbols.childrenKey] = []

      if (pluginMeta && pluginMeta.name && hemera._config.childLogger) {
        instance.log = hemera.log.child({ plugin: pluginMeta.name })
      }

      instance[Symbols.registeredPlugins] = Object.create(
        hemera[Symbols.registeredPlugins]
      )

      // inherit all extensions
      instance._extensionManager = ExtensionManager.build(
        hemera._extensionManager
      )

      // decorate root instance. All instances will have access
      instance.decorate = function() {
        hemera.decorate.apply(this._root, arguments)
        return instance
      }

      return instance
    }
  }

  /**
   *
   *
   * @memberof Hemera
   */
  _configureLogger() {
    const loggerOpts = {
      name: this._config.name,
      prettyPrint: this._config.prettyLog,
      level: this._config.logLevel
    }
    if (this._config.logger instanceof Stream) {
      this.log = Pino(loggerOpts, this._config.logger)
    } else if (this._config.logger) {
      this.log = this._config.logger
    } else {
      this.log = Pino(loggerOpts)
    }
  }

  /**
   *
   *
   * @memberof Hemera
   */
  _registerErrors() {
    for (let error in Hemera.errors) {
      Errio.register(Hemera.errors[error])
    }
  }
  /**
   *
   * @param {*} fn
   */
  setIdGenerator(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError(`IdGenerator must be a function`)
    }
    this._idGenerator = fn

    return this
  }
  /**
   *
   * @param {*} fn
   */
  setServerDecoder(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError(`ServerDecoder must be a function`)
    }
    this._serverDecoder = fn

    return this
  }

  /**
   *
   * @param {*} fn
   */
  setServerEncoder(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError(`ServerEncoder must be a function`)
    }
    this._serverEncoder = fn

    return this
  }
  /**
   *
   * @param {*} fn
   */
  setClientDecoder(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError(`ClientDecoder must be a function`)
    }
    this._clientDecoder = fn

    return this
  }

  /**
   *
   * @param {*} fn
   */
  setClientEncoder(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError(`ClientEncoder must be a function`)
    }
    this._clientEncoder = fn

    return this
  }

  /**
   *
   * @param {*} fn
   */
  setNotFoundPattern(pattern) {
    // check for use quick syntax for JSON objects
    if (typeof pattern === 'string') {
      pattern = TinySonic(pattern)
    }

    this._notFoundPattern = pattern

    return this
  }

  /**
   * Return the bloomrun instance
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  get router() {
    return this._router
  }

  /**
   * Return the heavy instance
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  get load() {
    return this._heavy.load
  }

  /**
   * Return the underlying NATS driver
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  get transport() {
    return this._transport.driver
  }

  /**
   * Return all registered topics
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  get topics() {
    return this._topics
  }

  /**
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  get notFoundPattern() {
    return this._notFoundPattern
  }

  /**
   *
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  get config() {
    return this._config
  }

  /**
   *
   *
   * @readonly
   *
   * @memberof Hemera
   */
  get errorDetails() {
    if (this._isServer) {
      return {
        app: this._config.name,
        isServer: this._isServer,
        pattern: this.trace$.method
      }
    } else {
      return {
        app: this._config.name,
        isServer: this._isServer,
        pattern: this.trace$.method
      }
    }
  }

  /**
   * Return all hemera errors
   *
   * @readonly
   * @static
   *
   * @memberOf Hemera
   */
  static get errors() {
    return Errors
  }

  /**
   * Create a custom super error object without to start hemera
   *
   * @readonly
   *
   * @memberOf Hemera
   */
  static createError(name) {
    if (typeof name !== 'string') {
      throw new Errors.HemeraError(`Error name must be a string`)
    }

    const ctor = SuperError.subclass(name)
    // Register the class with Errio.
    Errio.register(ctor)
    return ctor
  }
  /**
   * Add an onAdd handler
   *
   * @param {any} handler
   *
   * @memberOf Hemera
   */
  _addOnAddHandler(handler) {
    this._onAddHandlers.push(handler)
  }

  /**
   * Add an extension. Extensions are called in serie
   *
   * @param {any} type
   * @param {any} handler
   *
   * @memberOf Hemera
   */
  ext(type, handler) {
    if (typeof handler !== 'function') {
      throw new Errors.HemeraError('Extension handler must be a function')
    }

    if (type === 'onAdd') {
      this._addOnAddHandler(handler)
    } else if (type === 'onClose') {
      this.onShutdown(handler)
    } else {
      this._extensionManager.add(type, handler)
    }

    return this
  }

  /**
   *
   * @param {*} fn
   */
  setSchemaCompiler(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError('SchemaCompiler handler must be a function')
    }
    this._schemaCompiler = fn

    return this
  }

  /**
   *
   * @param {*} fn
   */
  setResponseSchemaCompiler(fn) {
    if (typeof fn !== 'function') {
      throw new Errors.HemeraError(
        'ResponseSchemaCompiler handler must be a function'
      )
    }
    this._responseSchemaCompiler = fn

    return this
  }

  /**
   * Exit the process
   *
   * @memberOf Hemera
   */
  fatal() {
    this.close(() => process.exit(1))
  }

  /**
   *
   *
   * @param {any} prop
   * @returns
   * @memberof Hemera
   */
  hasDecorator(prop) {
    return prop in this
  }

  /**
   * Decorate the root instance
   * Value is globaly accessible
   *
   * @param {any} prop
   * @param {any} value
   *
   * @memberOf Hemera
   */
  decorate(prop, value, deps) {
    if (prop in this) {
      throw new Errors.HemeraError('Decoration has been already added')
    }

    if (deps) {
      this._checkMemberDependencies(deps)
    }

    this[prop] = value

    return this
  }

  /**
   * Expose a value to the current instance
   *
   * @param {any} prop
   * @param {any} value
   *
   * @memberOf Hemera
   */
  expose(prop, value, deps) {
    if (prop in this) {
      throw new Errors.HemeraError('Exposition has been already added')
    }

    if (deps) {
      this._checkMemberDependencies(deps)
    }

    this[prop] = value

    return this
  }

  /**
   *
   *
   * @param {any} deps
   * @memberof Hemera
   */
  _checkMemberDependencies(deps) {
    for (let i = 0; i < deps.length; i++) {
      if (!(deps[i] in this)) {
        throw new Errors.HemeraError(`Missing member dependency '${deps[i]}'`)
      }
    }
  }

  /**
   *
   * @param {*} plugin
   */
  getPluginMeta(plugin) {
    return plugin[Symbol.for('plugin-meta')]
  }

  /**
   *
   *
   * @param {any} deps
   * @memberof Hemera
   */
  checkPluginDependencies(plugin) {
    const pluginMeta = this.getPluginMeta(plugin)
    if (!pluginMeta) {
      return
    }
    const dependencies = pluginMeta.dependencies
    if (!dependencies) {
      return
    }
    if (!Array.isArray(dependencies)) {
      throw new Errors.HemeraError(
        'Plugin dependencies must be an array of strings'
      )
    }
    dependencies.forEach(dependency => {
      if (this[Symbols.registeredPlugins].indexOf(dependency) === -1) {
        throw new Errors.HemeraError(
          `The dependency '${dependency}' is not registered`
        )
      }
    })
  }

  /**
   *
   *
   * @param {any} deps
   * @memberof Hemera
   */
  checkPluginDecorators(plugin) {
    const pluginMeta = this.getPluginMeta(plugin)
    if (!pluginMeta) {
      return
    }
    const decorators = pluginMeta.decorators
    if (!decorators) {
      return
    }
    if (!Array.isArray(decorators)) {
      throw new Errors.HemeraError(
        'Plugin decorators must be an array of strings'
      )
    }
    for (let i = 0; i < decorators.length; i++) {
      if (!(decorators[i] in this)) {
        throw new Errors.HemeraError(
          `The decorator dependency '${decorators[i]}' is not registered`
        )
      }
    }
  }

  /**
   *
   *
   * @param {any} plugin
   * @returns
   * @memberof Hemera
   */
  use(plugin, opts) {
    const pluginMeta = this.getPluginMeta(plugin)
    let pluginOpts = pluginMeta ? pluginMeta.options : {}
    pluginOpts = Object.assign({}, pluginOpts, opts)
    this.register(plugin, pluginOpts)
    return this._avvio
  }

  /**
   * Create a custom super error object in a running hemera instance
   *
   * @param {any} name
   * @returns
   *
   * @memberOf Hemera
   */
  createError(name) {
    const ctor = SuperError.subclass(name)
    // Register the class with Errio
    Errio.register(ctor)
    return ctor
  }

  /**
   *
   *
   * @param {Function} cb
   *
   * @memberOf Hemera
   */
  ready(cb) {
    if (this._isReady) {
      throw new Errors.HemeraError('Hemera was already bootstraped')
    }

    this._isReady = true

    this._transport.driver.on('error', err => {
      this.log.error(err, 'Could not connect to NATS!')
      this.log.error("NATS Code: '%s', Message: %s", err.code, err.message)

      // Exit only on connection issues.
      // Authorization and protocol issues don't lead to process termination
      if (natsConnCodes.indexOf(err.code) > -1) {
        // We have no NATS connection and can only gracefully shutdown hemera
        this.close()
      }
    })

    this._transport.driver.on('permission_error', err => {
      this.log.error(err, 'NATS permission error')
    })

    this._transport.driver.on('reconnect', () => {
      this.log.info('NATS reconnected!')
    })

    this._transport.driver.on('reconnecting', () => {
      this.log.warn('NATS reconnecting ...')
    })

    this._transport.driver.on('disconnect', () => {
      this.log.warn('NATS disconnected!')
    })

    // when nats was not able to reconnect or connection was closed due to other reasons
    // the process should die and restarted
    this._transport.driver.on('close', () => {
      const error = new Errors.HemeraError('NATS connection closed!')
      this.log.error(error)
      // when an 'error' handler was registered no error is thrown
      // but you have to handle it by yourself
      this.emit('error', error)
    })

    const ready = cb => {
      if (this._transport.driver.connected) {
        this.log.info('Connected!')
        this.bootstrap(cb)
      } else {
        this._transport.driver.on('connect', () => {
          this.log.info('Connected!')
          this.bootstrap(cb)
        })
      }
    }

    // callback style
    if (typeof cb === 'function') {
      ready(err => {
        if (err) {
          this.log.error(err)
          cb(err)
          return
        }
        cb()
      })
      return
    }

    // promise style
    return new Promise((resolve, reject) => {
      ready(err => {
        if (err) {
          this.log.error(err)
          reject(err)
          return
        }
        resolve()
      })
    })
  }

  /**
   *
   *
   * @param {any} err
   * @returns
   * @memberof Hemera
   */
  _attachHops(err) {
    const self = this

    if (err) {
      const errorDetails = {
        service: self.trace$.service,
        method: self.trace$.method,
        app: self._config.name,
        ts: Util.nowHrTime()
      }

      // collect hops
      if (err.hops) {
        err.hops.push(errorDetails)
      } else {
        err.hops = [errorDetails]
      }
    }

    return err
  }

  /**
   *  Attach one handler to the topic subscriber.
   *  With subToMany and maxMessages you control NATS specific behaviour.
   *
   * @param {any} addDefinition
   * @memberof Hemera
   */
  subscribe(addDefinition) {
    const self = this

    const topic = addDefinition.transport.topic
    const maxMessages = addDefinition.transport.maxMessages
    const queue = addDefinition.transport.queue || `queue.${topic}`
    const pubsub = addDefinition.transport.pubsub

    // avoid duplicate subscribers of the emit stream
    // we use one subscriber per topic
    if (self._topics.has(topic)) {
      return 0
    }

    let handler = (request, replyTo) => {
      // create new execution context
      let hemera = self.createContext()
      hemera._topic = topic
      hemera.request = new ServerRequest(request)
      hemera.response = new ServerResponse(replyTo)
      hemera.reply = new Reply(
        hemera.request,
        hemera.response,
        hemera,
        hemera.log
      )
      hemera._pattern = null
      hemera._isServer = true

      // represent the matched server action "add"
      hemera.matchedAction = null

      runExt(
        hemera._extensionManager.onServerPreRequest,
        serverExtIterator,
        hemera,
        err => hemera._onServerPreRequestCompleted(err)
      )
    }

    // standard pubsub with optional max messages
    if (pubsub) {
      return self._transport.subscribe(
        topic,
        {
          max: maxMessages
        },
        handler
      )
    } else {
      // queue group names allow load balancing of services
      return self._transport.subscribe(
        topic,
        {
          queue,
          max: maxMessages
        },
        handler
      )
    }
  }

  /**
   *
   * @param {any} extensionError
   * @param {any} value
   * @memberof Hemera
   */
  _onServerPreRequestCompleted(extensionError) {
    const self = this

    if (extensionError) {
      const internalError = new Errors.HemeraError(
        'onServerPreRequest extension',
        self.errorDetails
      ).causedBy(extensionError)
      self.log.error(internalError)
      self.emit('serverResponseError', extensionError)
      self.reply.isError = true
      self.reply.send(extensionError)
      return
    }

    // check if a handler is registered with this pattern
    if (self.matchedAction) {
      self.emit('serverPreHandler', self)
      if (self._extensionManager.onServerPreHandler.length) {
        runExt(
          self._extensionManager.onServerPreHandler,
          serverExtIterator,
          self,
          err => self._onServerPreHandlerCompleted(err)
        )
      } else {
        self._onServerPreHandlerCompleted()
      }
    } else {
      const internalError = new Errors.PatternNotFound(
        'No action found for this pattern',
        self.errorDetails
      )
      self.log.error(internalError)
      self.emit('serverResponseError', internalError)
      self.reply.isError = true
      self.reply.send(internalError)
    }
  }

  /**
   *
   *
   * @param {any} extensionError
   * @memberof Hemera
   */
  _onServerPreHandlerCompleted(extensionError) {
    const self = this

    if (extensionError) {
      const internalError = new Errors.HemeraError(
        'onServerPreHandler extension',
        self.errorDetails
      ).causedBy(extensionError)
      self.log.error(internalError)
      self.emit('serverResponseError', extensionError)
      self.reply.isError = true
      self.reply.send(extensionError)
      return
    }

    // action middleware
    self.matchedAction.run(self.request, self.reply, err =>
      self._afterMiddlewareHandler(err)
    )
  }

  /**
   *
   *
   * @param {any} err
   * @memberof Hemera
   */
  _afterMiddlewareHandler(err) {
    const self = this

    if (err) {
      const internalError = new Errors.HemeraError(
        'Action middleware',
        self.errorDetails
      ).causedBy(err)
      self.log.error(internalError)
      self.emit('serverResponseError', err)
      self.reply.isError = true
      self.reply.send(err)
      return
    }

    const result = self._processServerAction()

    const isPromise = result && typeof result.then === 'function'
    if (isPromise) {
      // avoid to create a seperate promise
      // eslint-disable-next-line promise/catch-or-return
      result.then(
        payload => self.reply.send(payload),
        err => {
          self.reply.isError = true
          self.reply.send(err)
        }
      )
    }
  }

  /**
   *
   *
   * @returns
   * @memberof Hemera
   */
  _processServerAction() {
    const self = this
    let action = self.matchedAction.action.bind(self)

    // if request type is 'pubsub' we don't have to reply back
    if (self.request.payload.request.type === 'pubsub') {
      action(self.request.payload.pattern)
      self.reply.send()
      return
    }

    return action(self.request.payload.pattern, (err, result) => {
      if (err) {
        self.reply.isError = true
        self.reply.send(err)
        return
      }
      self.reply.send(result)
    })
  }

  /**
   * Unsubscribe a topic or subscription id from NATS and Hemera
   *
   * @param {any} topic
   * @param {any} maxMessages
   * @returns
   *
   * @memberOf Hemera
   */
  remove(topic, maxMessages) {
    const self = this

    if (!topic) {
      let error = new Errors.HemeraError('The sid or topic name is required')
      self.log.error(error)
      throw error
    }

    // when sid was passed
    if (typeof topic === 'string') {
      // when topic name was passed
      const subId = self._topics.get(topic)

      if (subId) {
        self._transport.unsubscribe(subId, maxMessages)
        // we remove all subscription related to this topic
        this.cleanTopic(topic)
        return true
      }
    } else {
      self._transport.unsubscribe(topic, maxMessages)
      return true
    }

    return false
  }

  /**
   * Remove topic from list and remove all patterns from index
   *
   * @param {any} topic
   * @memberof Hemera
   */
  cleanTopic(topic) {
    // release topic so we can add it again
    this._topics.delete(topic)
    // remove pattern which belongs to the topic
    this.list().forEach(add => {
      if (add.pattern.topic === topic) {
        this.router.remove(add.pattern)
      }
    })
  }

  /**
   * The topic is subscribed on NATS and can be called from any client.
   *
   * @param {any} pattern
   * @param {any} cb
   *
   * @memberOf Hemera
   */
  add(pattern, cb) {
    if (!pattern) {
      let error = new Errors.HemeraError(
        'Pattern is required to define a server action'
      )
      this.log.error(error)
      throw error
    }

    // check for use quick syntax for JSON objects
    if (typeof pattern === 'string') {
      pattern = TinySonic(pattern)
    }

    // topic is needed to subscribe on a subject in NATS
    if (!pattern.topic) {
      let error = new Errors.HemeraError(
        'Topic is required and must be from type string',
        this.errorDetails
      )
      this.log.error(error)
      throw error
    }

    let schema = Util.extractSchema(pattern)
    let patternOnly = Util.cleanPattern(pattern)

    let addDefinition = new Add({
      schema: schema,
      pattern: patternOnly,
      transport: {
        topic: pattern.topic,
        pubsub: pattern.pubsub$,
        maxMessages: pattern.maxMessages$,
        queue: pattern.queue$
      }
    })

    if (cb) {
      // set callback
      addDefinition.action = cb
    }

    // Support full / token wildcards in subject
    // Convert nats wildcard tokens to RegexExp
    addDefinition.pattern.topic = Util.natsWildcardToRegex(patternOnly.topic)

    let handler = this._router.lookup(addDefinition.pattern)

    // check if pattern is already registered
    if (this._config.bloomrun.lookupBeforeAdd && handler) {
      let error = new Errors.HemeraError('Pattern is already in use', {
        pattern: addDefinition.pattern
      })

      this.log.error(
        {
          pattern: addDefinition.pattern
        },
        error
      )
      throw error
    }

    // check for invalid topic subscriptions
    // it's not possible to susbcribe to the same topic with different transport options
    // because we use one subscription for the topic
    const def = this._checkForTransportCollision(addDefinition)
    if (def) {
      this.log.error(
        'Topic is already registered with special transport options. Please use a different topic name.',
        Util.pattern(def.pattern)
      )
      throw new Errors.HemeraError(
        'Topic is already registered with special transport options'
      )
    }

    // add to bloomrun
    this._router.add(patternOnly, addDefinition)

    this.log.info(patternOnly, 'Server action added')

    const sid = this.subscribe(addDefinition)

    if (sid > 0) {
      addDefinition.sid = sid
      this._topics.set(addDefinition.pattern.topic, sid)
    }

    this._runOnAddHandler(addDefinition)

    return addDefinition
  }
  /**
   * Run all onAdd handlers in serie
   * options
   *
   * @param {any} addDefinition
   * @memberof Hemera
   */
  _runOnAddHandler(addDefinition) {
    runExt(
      this._onAddHandlers,
      (fn, state, next) => {
        fn(addDefinition)
        next()
      },
      this,
      () => {}
    )
  }
  /**
   * Check if a topic was already registered with different transport
   * options
   *
   * @param {any} addDefinition
   * @memberof Hemera
   */
  _checkForTransportCollision(addDefinition) {
    const mT2 = addDefinition.transport
    for (const def of this._router) {
      const mT1 = def.transport
      // looking for another pattern with same topic but
      // different transport options
      if (addDefinition.pattern.topic === def.pattern.topic) {
        if (
          mT1.maxMessages !== mT2.maxMessages ||
          mT1.queue !== mT2.queue ||
          mT1.pubsub !== mT2.pubsub
        ) {
          return def
        }
      }
    }
    return null
  }

  /**
   *
   *
   * @param {any} response
   *
   * @memberof Hemera
   */
  _sendRequestHandler(response) {
    const self = this

    if (response.code && response.code === NATS.REQ_TIMEOUT) {
      self._timeoutHandler()
      return
    }

    const res = self._clientDecoder(response)
    self.response.payload = res.value
    self.response.error = res.error

    // decoding error
    if (self.response.error) {
      let internalError = new Errors.ParseError(
        'Client payload decoding',
        self.errorDetails
      ).causedBy(self.response.error)
      self.log.error(internalError)
      self.emit('clientResponseError', self.response.error)
      self._execute(self.response.error)
      return
    }

    runExt(
      self._extensionManager.onClientPostRequest,
      clientExtIterator,
      self,
      err => self._onClientPostRequestCompleted(err)
    )
  }

  /**
   *
   *
   * @param {any} extensionError
   * @memberof Hemera
   */
  _onClientPostRequestCompleted(extensionError) {
    const self = this

    if (extensionError) {
      let error = self.getRootError(extensionError)
      const internalError = new Errors.HemeraError(
        'onClientPostRequest extension',
        self.errorDetails
      ).causedBy(extensionError)
      self.log.error(internalError)
      self.emit('clientResponseError', extensionError)
      self._execute(error)
      return
    }

    if (self.response.payload.error) {
      let error = Errio.fromObject(self.response.payload.error)
      self.emit('clientResponseError', error)
      self._execute(error)
      return
    }

    // call act handler
    self._execute(null, self.response.payload.result)
  }

  /**
   * Start an action.
   *
   * @param {any} pattern
   * @param {any} cb
   *
   * @memberOf Hemera
   */
  act(pattern, cb) {
    if (!pattern) {
      let error = new Errors.HemeraError(
        'Pattern is required to start a request'
      )
      this.log.error(error)
      throw error
    }

    // check for use quick syntax for JSON objects
    if (typeof pattern === 'string') {
      pattern = TinySonic(pattern)
    }

    // topic is needed to subscribe on a subject in NATS
    if (!pattern.topic) {
      let error = new Errors.HemeraError(
        'Topic is required and must be from type string',
        this.errorDetails
      )
      this.log.error(error)
      throw error
    }

    // create new execution context
    let hemera = this.createContext()
    hemera._pattern = pattern
    hemera._parentContext = this
    hemera._cleanPattern = Util.cleanFromSpecialVars(pattern)
    hemera._isServer = false
    hemera._execute = null

    hemera.response = new ClientResponse()
    hemera.request = new ClientRequest(pattern)
    hemera.sid = 0

    if (cb) {
      hemera._execute = cb.bind(hemera)
      runExt(
        hemera._extensionManager.onClientPreRequest,
        clientExtIterator,
        hemera,
        err => hemera._onClientPreRequestCompleted(err)
      )
    } else {
      const evaluateResult = new Promise((resolve, reject) => {
        hemera._execute = (err, result) => {
          if (err) {
            reject(err)
          } else {
            resolve(result)
          }
        }
      })

      runExt(
        hemera._extensionManager.onClientPreRequest,
        clientExtIterator,
        hemera,
        err => hemera._onClientPreRequestCompleted(err)
      )

      return evaluateResult.then(resp => {
        return {
          context: hemera,
          data: resp
        }
      })
    }
  }

  /**
   *
   *
   * @param {any} err
   * @returns
   * @memberof Hemera
   */
  getRootError(err) {
    let error = null
    if (err instanceof SuperError) {
      error = err.rootCause || err.cause || err
    } else {
      error = err
    }
    return error
  }

  /**
   *
   *
   * @param {any} err
   * @memberof Hemera
   */
  _onClientPreRequestCompleted(err) {
    const self = this
    let m = self._clientEncoder(self._message)

    self.request.payload = m.value

    // encoding issue
    if (m.error) {
      self.request.payload = null
      self.request.error = m.error
      let error = new Errors.ParseError('Client payload encoding').causedBy(
        m.error
      )
      self.log.error(error)
      self.emit('clientResponseError', m.error)
      self._execute(m.error)
      return
    }

    if (err) {
      let error = self.getRootError(err)
      self.request.payload = null
      self.request.error = error
      const internalError = new Errors.HemeraError(
        'onClientPreRequest extension'
      ).causedBy(err)
      self.log.error(internalError)
      self.emit('clientResponseError', error)
      self._execute(error)
      return
    }

    // use simple publish mechanism instead of request/reply
    if (self._pattern.pubsub$ === true) {
      self._transport.send(self._pattern.topic, self.request.payload, err =>
        self._execute(err)
      )
    } else {
      const optOptions = {
        timeout: self._pattern.timeout$ || self.config.timeout,
        // default is request-reply semantic but we can assign -1 (no limit)
        // but the user has to unsubscribe it manually
        max: 1
      }
      // limit on the number of responses the requestor may receive
      // support maxMessages$ -1
      if (
        typeof self._pattern.maxMessages$ === 'number' ||
        self._pattern.expectedMessages$ > 0
      ) {
        // we can't receive more messages than "expected" messages
        // the inbox is closed automatically
        optOptions.max =
          self._pattern.expectedMessages$ || self._pattern.maxMessages$
      }
      // send request
      self.sid = self._transport.sendRequest(
        self._pattern.topic,
        self.request.payload,
        optOptions,
        resp => self._sendRequestHandler(resp)
      )

      // create timeout handler only with a combination of expected msg
      // the default timeout handler is created by NATS client
      if (self._pattern.expectedMessages$ > 0) {
        self._handleTimeout()
      }
    }
  }
  /**
   * Handle the timeout when a pattern could not be resolved. Can have different reasons:
   * - No one was connected at the time (service unavailable)
   * - Service is actually still processing the request (service takes too long)
   * - Service was processing the request but crashed (service error)
   *
   * @memberOf Hemera
   */
  _handleTimeout() {
    const self = this

    self._transport.timeout(
      self.sid,
      self._pattern.timeout$ || this._config.timeout,
      self._pattern.expectedMessages$,
      () => this._timeoutHandler()
    )
  }

  /**
   *
   * @memberof Hemera
   */
  _timeoutHandler() {
    const self = this

    const error = new Errors.TimeoutError('Client timeout', self.errorDetails)
    self.log.error(error)
    self.response.error = error
    self.emit('clientResponseError', error)

    runExt(
      self._extensionManager.onClientPostRequest,
      clientExtIterator,
      self,
      err => self._onClientTimeoutPostRequestCompleted(err)
    )
  }

  /**
   *
   *
   * @param {any} err
   * @memberof Hemera
   */
  _onClientTimeoutPostRequestCompleted(err) {
    const self = this

    if (err) {
      let error = self.getRootError(err)
      const internalError = new Errors.HemeraError(
        'onClientPostRequest extension'
      ).causedBy(err)
      self.log.error(internalError)
      self.response.error = error
      self.emit('clientResponseError', error)
    }

    self._execute(self.response.error)
  }

  /**
   * Create new instance of hemera but based on the current prototype
   * so we are able to create a scope per act without lossing the reference to the core api.
   *
   * @returns
   *
   * @memberOf Hemera
   */
  createContext() {
    return Object.create(this)
  }

  /**
   * Return the list of all registered actions
   *
   * @param {any} pattern
   * @param {any} options
   * @returns
   *
   * @memberOf Hemera
   */
  list(pattern, options) {
    return this._router.list(pattern, options)
  }

  /**
   * Remove all registered pattern and release topic from NATS
   *
   * @memberof Hemera
   */
  removeAll() {
    for (var topic of this._topics.keys()) {
      this.remove(topic)
    }
  }

  /**
   *
   *
   * @memberof Hemera
   */
  close(func) {
    // 2. clean hemera
    this.onShutdown((instance, done) => {
      this._heavy.stop()
      done()
    })

    // 1. clean nats
    this.onShutdown((instance, done) => {
      this.removeAll()
      this._transport.flush(() => {
        this._transport.close()
        done()
      })
    })

    return this.shutdown(func)
  }
}

module.exports = Hemera