restream/src/Replaceable.js

import Stream, { Transform } from 'stream'
import cleanStack from '@artdeco/clean-stack'
import { collect } from 'catchment'
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
    this._options = options
  }

  /**
   * Creates a new replaceable to replace the given string, buffer or stream using the rules of the current stream. Calling `brake` will also set `_broke` on the parent stream.
   * @param {string|!Buffer|!Stream} input The input to replace.
   * @param {!Object} [context] The context to assign to the new replaceable.
   */
  async replace(input, context) {
    const replaceable = new Replaceable(this.rules, this._options)
    if (context) Object.assign(replaceable, context)
    const res = await Replaceable.replace(replaceable, input)
    if (replaceable._broke) this.brake()
    if (context) Object.keys(context).forEach(key => {
      context[key] = replaceable[key]
    })
    return res
  }
  /**
   * The method to end the stream with given data, and collect the result.
   * @param {!Replaceable} replaceable
   * @param {string|!Buffer|!Stream} input The data or stream to end the _Replaceable_ with.
   */
  static async replace(replaceable, input) {
    if (input instanceof Stream) {
      input.pipe(replaceable)
    } else {
      replaceable.end(input)
    }
    /** @type {string} */
    const data = await collect(replaceable)
    return data
  }

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

  /**
   * @param {string|!Buffer} chunk The incoming chunk, or an object if the stream was started in object mode.
   */
  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) {
              if (promises.length)
                return promises.push(Promise.resolve(match))
              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
 */