<commit_before>"use strict";

var Q = require('q');
var PassThrough = require('stream').PassThrough;
var persistence = require('./persistence');

var examineKey = function (key) {
    if (persistence.isValidKey(key) === false) {
        throw new Error('Unsupported key type');
    }
};

var examineKeyList = function (list) {
    if (Array.isArray(list)) {
        var areValuesOk = list.reduce(function (previousValue, currentValue) {
            if (previousValue === false || persistence.isValidKey(currentValue) === false) {
                return false;
            }
            return true;
        }, true);
        if (areValuesOk) {
            return;
        }
    }
    throw new Error('Malformed array of keys');
};

module.exports.create = function (storageDirPath, callback) {
    
    // ---------------------------------------------
    // Initialization
    
    var readyToWork = false;
    callback = callback || function () {};
    
    var pers = persistence.create(storageDirPath, function (err) {
        if (err) {
            callback(err);
        } else {
            readyToWork = true;
            callback();
            runNextTask();
        }
    });
    
    // ---------------------------------------------
    // Queue of tasks
    
    var tasks = [];
    var waitingForIdle = [];
    var runningTask = null;
    
    var runNextTask = function () {
        if (!readyToWork || runningTask !== null) {
            return;
        }
        if (tasks.length === 0) {
            // All tasks executed, so entering "idle" mode.
            informAllWaitingForIdle();
            return;
        }
        
        runningTask = tasks.shift();
        var asyncInterface = runningTask.fn.apply(null, runningTask.args);
        switch (runningTask.fn.asyncInterfaceType) {
            case 'stream':
                asyncInterface.pipe(runningTask.stream);
                asyncInterface.on('end', function () {
                    runningTask = null;
                    runNextTask();
                });
                break;
            case 'promise':
                asyncInterface.then(function () {
                    runningTask.deferred.resolve.apply(null, arguments);
                })
                .catch(function () {
                    runningTask.deferred.reject.apply(null, arguments);
                })
                .finally(function () {
                    runningTask = null;
                    runNextTask();
                });
                break;
        }
    };
    
    var addToQueue = function (fn, args) {
        var ret;
        var task = {
            fn: fn,
            args: args,
        };
        
        if (fn.asyncInterfaceType === 'stream') {
            task.stream = new PassThrough({ objectMode: true });
            ret = task.stream;
        } else {
            task.deferred = Q.defer();
            ret = task.deferred.promise;
        }
        
        tasks.push(task);
        
        runNextTask();
        
        return ret;
    };
    
    var informAllWaitingForIdle = function () {
        while (waitingForIdle.length > 0) {
            waitingForIdle.pop().resolve();
        }
    };
    
    // ---------------------------------------------
    // API
    
    var get = function (key) {
        examineKey(key);
        return addToQueue(pers.get, [key]);
    };
    
    var getMany = function (keys) {
        examineKeyList(keys);
        return addToQueue(pers.getMany, [keys]);
    };
    
    var getAll = function () {
        return addToQueue(pers.getAll, []);
    };
    
    var set = function (key, value) {
        examineKey(key);
        return addToQueue(pers.set, [key, value]);
    };
    
    var del = function (key) {
        examineKey(key);
        return addToQueue(pers.del, [key]);
    };
    
    var whenIdle = function () {
        var deferred = Q.defer();
        waitingForIdle.push(deferred);
        runNextTask();
        return deferred.promise; 
    };
    
    return {
        set: set,
        get: get,
        getMany: getMany,
        getAll: getAll,
        delete: del,
        whenIdle: whenIdle,
    };
};<commit_msg>Migrated code style of core<commit_after>const Q = require('q');
const PassThrough = require('stream').PassThrough;
const persistence = require('./persistence');

const examineKey = (key) => {
  if (persistence.isValidKey(key) === false) {
    throw new Error('Unsupported key type');
  }
};

const examineKeyList = (list) => {
  if (Array.isArray(list)) {
    const areValuesOk = list.reduce((previousValue, currentValue) => {
      if (previousValue === false || persistence.isValidKey(currentValue) === false) {
        return false;
      }
      return true;
    }, true);
    if (areValuesOk) {
      return;
    }
  }
  throw new Error('Malformed array of keys');
};

module.exports.create = (storageDirPath, cb) => {
  const callback = cb || function () {};
  const tasks = [];
  const waitingForIdle = [];
  let readyToWork = false;
  let runningTask = null;

  const informAllWaitingForIdle = function () {
    while (waitingForIdle.length > 0) {
      waitingForIdle.pop().resolve();
    }
  };

  const runNextTask = function () {
    if (!readyToWork || runningTask !== null) {
      return;
    }
    if (tasks.length === 0) {
      // All tasks executed, so entering "idle" mode.
      informAllWaitingForIdle();
      return;
    }

    runningTask = tasks.shift();
    const asyncInterface = runningTask.fn.apply(null, runningTask.args);
    switch (runningTask.fn.asyncInterfaceType) {
      case 'stream':
        asyncInterface.pipe(runningTask.stream);
        asyncInterface.on('end', () => {
          runningTask = null;
          runNextTask();
        });
        break;
      case 'promise':
        asyncInterface.then(function () {
          runningTask.deferred.resolve.apply(null, arguments);
        })
        .catch(function () {
          runningTask.deferred.reject.apply(null, arguments);
        })
        .finally(() => {
          runningTask = null;
          runNextTask();
        });
        break;
      default:
        throw new Error('Unknown interface');
    }
  };

  const addToQueue = function (fn, args) {
    let ret;
    const task = {
      fn,
      args,
    };

    if (fn.asyncInterfaceType === 'stream') {
      task.stream = new PassThrough({ objectMode: true });
      ret = task.stream;
    } else {
      task.deferred = Q.defer();
      ret = task.deferred.promise;
    }

    tasks.push(task);

    runNextTask();

    return ret;
  };

  // ---------------------------------------------
  // Initialization

  const pers = persistence.create(storageDirPath, (err) => {
    if (err) {
      callback(err);
    } else {
      readyToWork = true;
      callback();
      runNextTask();
    }
  });

  // ---------------------------------------------
  // API

  const get = function (key) {
    examineKey(key);
    return addToQueue(pers.get, [key]);
  };

  const getMany = function (keys) {
    examineKeyList(keys);
    return addToQueue(pers.getMany, [keys]);
  };

  const getAll = function () {
    return addToQueue(pers.getAll, []);
  };

  const set = function (key, value) {
    examineKey(key);
    return addToQueue(pers.set, [key, value]);
  };

  const del = function (key) {
    examineKey(key);
    return addToQueue(pers.del, [key]);
  };

  const whenIdle = function () {
    const deferred = Q.defer();
    waitingForIdle.push(deferred);
    runNextTask();
    return deferred.promise;
  };

  return {
    set,
    get,
    getMany,
    getAll,
    delete: del,
    whenIdle,
  };
};
{'fraction_non_alphanumeric': 0.15586621374444734, 'fraction_numerical': 0.0005226025607525476, 'mean_word_length': 1.669107391910739, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 12, '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': '703474', 'n_tokens_mistral': 2190, 'n_tokens_neox': 2158, 'n_words': 470}