restream/src/Replaceable.js

import { Transform } from 'stream'
import cleanStack from '@artdeco/clean-stack'
import { checkRule, hideStack } from './lib'

export default class Replaceable extends Transform {
  /**
   * Replaceable class that extends Transform and pushes data when it's done replacing each incoming chunk. If the replacement is passed as a function, it will work in the same way as the replacer for `string.replace` method (https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/String/replace), taking the `match` as the first argument, and matched `p1`, `p2`, _etc_ parameters as following arguments. The replacer can also be an async function.
   * @param {!(_restream.Rule|Array<!_restream.Rule>)} rules A single replacement rule, or multiple rules.
   * @param {!stream.TransformOptions} [options] The options for the transform stream.
   * @example
   *
```
// markdown __ to html <em> implementation
const stream = replaceStream({
  re: /__(\S+)__/g,
  replacement(match, p1) {
    return `<em>${p1}</em>`
  },
})
```
   */
  constructor(rules, options) {
    super(options)
    const re = Array.isArray(rules) ? rules : [rules]
    const fr = re.filter(checkRule)
    this.rules = fr
    /**
     * Whether the _Replaceable_ will not apply any more rules.
     * @type {boolean}
     */
    this._broke = false
  }

  /**
   * Stop executing further after the current rule.
   */
  brake() {
    this._broke = true
  }

  async reduce(chunk) {
    /** @type {string} */
    const s = await this.rules.reduce(async (acc, { re, replacement }) => {
      /** @type {string} */
      let string = await acc
      if (this._broke) return string

      if (typeof replacement == 'string') {
        string = string.replace(re, replacement)
      } else {
        const promises = []
        let commonError
        const t = string.replace(re, (match, ...args) => {
          commonError = new Error()
          try {
            if (this._broke) return match
            const p = replacement.call(this, match, ...args)
            if (p instanceof Promise) {
              promises.push(p)
            }
            return p
          } catch (e) { // hide stack for sync stack traces
            hideStack(commonError, e)
          }
        })
        if (promises.length) {
          try { // hide stack only for when throw happens before awaits
            const data = await Promise.all(promises)
            string = string.replace(re, () => data.shift())
          } catch (e) {
            hideStack(commonError, e)
          }
        } else {
          string = t
        }
      }
      return string
    }, `${chunk}`)

    return s
  }
  /**
   * @suppress {checkTypes}
   * @returns {!Promise}
   */
  async _transform(chunk, _, next) {
    try {
      const s = await this.reduce(chunk)
      this.push(s)
      next()
    } catch (e) {
      const s = cleanStack(e.stack)
      e.stack = s
      next(e)
    }
  }
}

/**
 * The class for when serial execution of asynchronous replacements withing the same rule are needed.
 */
export class SerialAsyncReplaceable extends Replaceable {
  /**
   * @param {!(_restream.Rule|Array<_restream.Rule>)} rules
   */
  constructor(rules) {
    super(rules)
    this.promise = Promise.resolve()
  }
  addItem(fn) {
    const pp = this.promise.then(fn)
    this.promise = pp
    return pp
  }
}

/**
 * @suppress {nonStandardJsDocs}
 * @typedef {import('..').Rule} _restream.Rule
 */
/**
 * @suppress {nonStandardJsDocs}
 * @typedef {import('..').AsyncReplacer} _restream.AsyncReplacer
 */
/**
 * @suppress {nonStandardJsDocs}
 * @typedef {import('..').Replacer} _restream.Replacer
 */
/**
 * @suppress {nonStandardJsDocs}
 * @typedef {import('stream').TransformOptions} stream.TransformOptions
 */