<commit_before>let noop = () => {};

/**
 * 生成一个处理数据的 Pipeline 包装
 *
 * @param  {AnyType} value      待处理的值
 * @param  {Object} prePipeline
 *     ## 定义 Pipeline 开始前进的操作。
 *     ```
 *     {
 *       name: [String],  // Default 'Start Pipeline'
 *       mixins: [Array]
 *      }
 *     ```
 * @return {Object}        一个包含 flow 和 finish 属性的对象。 两个属性均为函数。
 *                         两个属性均接受一个函数作为，该函数接受一个数据，并返回一个新的数据
 */
export function pipelineWrap (input, prePipeline={}) {
  let _preP = prePipeline;
  _preP.name = _preP.name || 'Finish Pipeline';
  _preP.mixins = _preP.mixins || [];

  let _pre = globalExecMixins(_preP.mixins, {name: _preP.name, value: input});
  let _output = _pre.value;
  let _pipeCount = 0;
  let _blocking = false;
  let _asyncPipeQueue = [/*{name, pipe}*/];
  let _syncPipesAfterAsync = new Map(); // {[Async{name, pipe}] => [Sync{name, pipe}]}

  let _handleSuccFinish = noop;
  let _handleErrFinish = noop;

  let wrapper = {
    /**
     * 同步执行函数
     * @param  {Object} pipeObj [description]
     * @return {[type]}         [description]
     */
    flow(pipeObj) {
      pipeObj = _preparePipeObj(pipeObj, 'flow');
      _addSyncPipe(pipeObj);
      return wrapper;
    },

    flowAsync(pipeObj) {
      pipeObj = _preparePipeObj(pipeObj, 'flowAsync');

      _asyncPipeQueue.push(pipeObj);
      _syncPipesAfterAsync.set(pipeObj, []);
      _blocking || _execAsyncPipe(_output);
      return wrapper;
    },

    mapFlow(pipeObj) {
      pipeObj = _preparePipeObj(pipeObj, 'mapFlow');
      pipeObj.type = 'map';
      _addSyncPipe(pipeObj);
      return wrapper;<commit_msg>add docs<commit_after>let noop = () => {};

/**
 * 生成一个处理数据的 Pipeline 包装
 *
 * @param  {AnyType} value      待处理的值
 * @param  {Object} prePipeline
 *     ## 定义 Pipeline 开始前进的操作。
 *     ```
 *     {
 *       name: [String],  // Default 'Start Pipeline'
 *       mixins: [Array]
 *      }
 *     ```
 * @return {Object}        一个包含 flow 和 finish 属性的对象。 两个属性均为函数。
 *                         两个属性均接受一个函数作为，该函数接受一个数据，并返回一个新的数据
 */
export function pipelineWrap (input, prePipeline={}) {
  let _preP = prePipeline;
  _preP.name = _preP.name || 'Finish Pipeline';
  _preP.mixins = _preP.mixins || [];

  let _pre = globalExecMixins(_preP.mixins, {name: _preP.name, value: input});
  let _output = _pre.value;
  let _pipeCount = 0;
  // 执行异步 Pipe 时设置为 true
  let _blocking = false;
  // 异步 Pipe 队列。当 Pipe 执行完成后才会从队列中移除
  let _asyncPipeQueue = [/*{name, pipe}*/];
  // 异步 Pipe 之后的同步 Pipe 队列
  let _syncPipesAfterAsync = new Map(); // {[Async{name, pipe}] => [Sync{name, pipe}]}

  let _handleSuccFinish = noop;
  let _handleErrFinish = noop;

  let wrapper = {
    /**
     * 同步执行函数
     * @param  {Object} pipeObj [description]
     * @return {[type]}         [description]
     */
    flow(pipeObj) {
      pipeObj = _preparePipeObj(pipeObj, 'flow');
      _addSyncPipe(pipeObj);
      return wrapper;
    },

    flowAsync(pipeObj) {
      pipeObj = _preparePipeObj(pipeObj, 'flowAsync');

      _asyncPipeQueue.push(pipeObj);
      _syncPipesAfterAsync.set(pipeObj, []);
      _blocking || _execAsyncPipe(_output);
      return wrapper;
    },

    mapFlow(pipeObj) {
      pipeObj = _preparePipeObj(pipeObj, 'mapFlow');
      pipeObj.type = 'map';
      _addSyncPipe(pipeObj);
      return wrapper;
{'fraction_non_alphanumeric': 0.1409090909090909, 'fraction_numerical': 0.0006060606060606061, 'mean_word_length': 2.655592469545958, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '1962906', 'n_tokens_mistral': 1287, 'n_tokens_neox': 1321, 'n_words': 244}