<commit_before>export const FN_WAIT = 'wait'
export const FN_INPUT = 'input'
export const FN_OUTPUT = 'output'
export const FN_THROTTLE = 'throttle'


/**
 * Put function and its dependencies to the pipeline.
 *
 * @param  {Function|String|Number|Array} fn
 *         - Function: The pipe function
 *         - String: Name of function which could be found in dependencies.
 *         - Number: Number of miliseconds to throttle the pipeline.
 *         - Array: A array which contains both `fn` (first) and `deps` (rest).
 * @param  {Array|String}   ...input String or array of names of dependencies.
 * @return {Pipeline}       Instance of this Pipeline.
 */
export default function createPipe(pipeline, fn, input, output) {
  var fnType = typeof fn

  if ('number' === fnType) {
    return createThrottlePipe(fn)
  } else if ('string' === fnType) {
    switch (fn) {
      case FN_INPUT:
        // .pipe('input', ['input1', 'input2'])
        inputIsRequired(input, fnType)
        return createInputPipe(input)
      case FN_OUTPUT:
        // .pipe('output', ['result'])
        inputIsRequired(input, fnType)
        return createOutputPipe(input)
      case FN_THROTTLE:
        // .pipe('throttle', 300)
        inputIsRequired(input, fnType)
        return createThrottlePipe(input)
      case FN_WAIT:
        // .pipe('wait', 300)
        inputIsRequired(input, fnType)
        return createWaitPipe(input)
      default:
        return createInjectionPipe(fn, input, output)
    }
  } else if (true === fn.instanceOfAlfaPipeline) {
    fn = fn.toPipe(pipeline.getInjector())
  }

  return buildPipe(fn, input, output)
}

function inputIsRequired(input, fnType) {
  if (!input && 0 !== input)
    throw new Error(`'input' is required for ${fnType} pipe.`)
}

function normalizeStringInput(input, fnType) {
  if ('string' === typeof input)
    input = [input]

  if (!Array.isArray(input) || !input.every(item => 'string' === typeof item))
    throw new Error(`'${fnType}' pipe requires string or array of strings as input.`)

  return input
}

function createInputPipe(input) {
  input = normalizeStringInput(input)

  return {
    fn: function inputPipe(args, store, rawStore) {
      input.forEach((item, idx) => {
        if (item)
          rawStore[item] = args[idx]
      })
    }
  }
}

function createOutputPipe(input) {
  input = normalizeStringInput(input)

  return {
    fn: function outputPipe(args, store, rawStore) {
      const output = {}
      input.forEach(item => {
        if (item)
          output[item] = rawStore[item]
      })
      store.set(output)
    }
  }
}

function createWaitPipe(msec) {
}

function createThrottlePipe(msec) {
  // Generate a throttle function and push to pipes
  var timestamp
  return {
    fn: function throttlePipe() {
      var now = new Date()
      if (!timestamp || now - timestamp > msec) {<commit_msg>Wait pipe.<commit_after>export const FN_WAIT = 'wait'
export const FN_INPUT = 'input'
export const FN_OUTPUT = 'output'
export const FN_THROTTLE = 'throttle'


/**
 * Put function and its dependencies to the pipeline.
 *
 * @param  {Function|String|Number|Array} fn
 *         - Function: The pipe function
 *         - String: Name of function which could be found in dependencies.
 *         - Number: Number of miliseconds to throttle the pipeline.
 *         - Array: A array which contains both `fn` (first) and `deps` (rest).
 * @param  {Array|String}   ...input String or array of names of dependencies.
 * @return {Pipeline}       Instance of this Pipeline.
 */
export default function createPipe(pipeline, fn, input, output) {
  var fnType = typeof fn

  if ('number' === fnType) {
    return createThrottlePipe(fn)
  } else if ('string' === fnType) {
    switch (fn) {
      case FN_INPUT:
        // .pipe('input', ['input1', 'input2'])
        inputIsRequired(input, fnType)
        return createInputPipe(input)
      case FN_OUTPUT:
        // .pipe('output', ['result'])
        inputIsRequired(input, fnType)
        return createOutputPipe(input)
      case FN_THROTTLE:
        // .pipe('throttle', 300)
        inputIsRequired(input, fnType)
        return createThrottlePipe(input)
      case FN_WAIT:
        // .pipe('wait', 300)
        inputIsRequired(input, fnType)
        return createWaitPipe(input)
      default:
        return createInjectionPipe(fn, input, output)
    }
  } else if (true === fn.instanceOfAlfaPipeline) {
    fn = fn.toPipe(pipeline.getInjector())
  }

  return buildPipe(fn, input, output)
}

function inputIsRequired(input, fnType) {
  if (!input && 0 !== input)
    throw new Error(`'input' is required for ${fnType} pipe.`)
}

function normalizeStringInput(input, fnType) {
  if ('string' === typeof input)
    input = [input]

  if (!Array.isArray(input) || !input.every(item => 'string' === typeof item))
    throw new Error(`'${fnType}' pipe requires string or array of strings as input.`)

  return input
}

function createInputPipe(input) {
  input = normalizeStringInput(input)

  return {
    fn: function inputPipe(args, store, rawStore) {
      input.forEach((item, idx) => {
        if (item)
          rawStore[item] = args[idx]
      })
    }
  }
}

function createOutputPipe(input) {
  input = normalizeStringInput(input)

  return {
    fn: function outputPipe(args, store, rawStore) {
      const output = {}
      input.forEach(item => {
        if (item)
          output[item] = rawStore[item]
      })
      store.set(output)
    }
  }
}

function createWaitPipe(msec) {
  return {
    fn: function waitPipe(next) {
      setTimeout(function() {
        next()
      }, msec)
    },
    input: ['next']
  }
}

function createThrottlePipe(msec) {
  // Generate a throttle function and push to pipes
  var timestamp
  return {
    fn: function throttlePipe() {
      var now = new Date()
      if (!timestamp || now - timestamp > msec) {
{'fraction_non_alphanumeric': 0.12327835402142492, 'fraction_numerical': 0.0030607039619112396, 'mean_word_length': 2.814526588845655, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1027918', 'n_tokens_mistral': 1862, 'n_tokens_neox': 1812, 'n_words': 512}