<commit_before>var redtape = require('redtape'),
    bytewise = require('bytewise'),
    level = require('level-test')({ keyEncoding: bytewise, valueEncoding: 'json' })
    after = require('after'),
    range = require('range'),
    uniq = require('lodash.uniq'),
    queue = require('..');

var it = redtape({
  beforeEach: function (cb) {
    var db = level('test');
    cb(null, db);
  },
  afterEach: function (db, cb) {
    db.close(cb);
  }
});

it('should be able to put and read the queue', 3, function(t, db) {
  var q = queue(db);
  var data = {
    id: 1,
    name: 'Eugene',
    value: 42
  };
  q.queue.push(data, read);
  function read(err) {
    t.notOk(err);
    q.queue.read(function (err, value) {
      t.notOk(err);
      t.deepEquals(data, value);
    });
  }
});

it('reading should dequeue the item', 4, function(t, db) {
  var q = queue(db);
  var data = {
    id: 1,
    name: 'Eugene',
    value: 42
  };
  q.queue.push(data, read);
  function read(err) {
    t.notOk(err);
    q.queue.read(function (err, value, key) {
      t.notOk(err);
      t.deepEquals(data, value);
      get(key);
    });
  }

  function get(key) {
    q.get(key, check);
  }

  function check(err, value) {
    t.ok(err);
  }
});

it('reading should be atomic', 1, function(t, db) {
  var q = queue(db);
  var next = after(5, dequeue);
  range(0, 5).forEach(function (i) {
    q.queue.push({
      id: i,
      name: 'Name ' + i,
      value: 42 + i
    }, next);
  });

  var results = [];
  function dequeue() {
    var next = after(5, check);
    range(0, 5).forEach(function (i) {
      q.queue.read(function (err, value, key) {
        results.push(value);
        next();
      });
    });
  }

  function prop(name) {
    return function (obj) {
      return obj[name];
    };
  }

  function check() {
    var hits = uniq(results.map(prop('id')));
    t.equal(hits.length, 5);
  }
});

it('reading should block until data is there', 3, function(t, db) {
  var q = queue(db);
  var data = {
    id: 1,
    name: 'Eugene',
    value: 42
  };
  q.queue.read(function (err, value) {
    t.notOk(err);
    t.deepEquals(data, value);
    t.ok(Date.now() - now >= 20);
  });
  var now = Date.now();
  setTimeout(function () {
    q.queue.push(data);
  }, 20);
});<commit_msg>custom prioritization<commit_after>var redtape = require('redtape'),
    bytewise = require('bytewise'),
    level = require('level-test')({ keyEncoding: bytewise, valueEncoding: 'json' })
    after = require('after'),
    range = require('range'),
    uniq = require('lodash.uniq'),
    queue = require('..');

var it = redtape({
  beforeEach: function (cb) {
    var db = level('test');
    cb(null, db);
  },
  afterEach: function (db, cb) {
    db.close(cb);
  }
});

function prop(name) {
  return function (obj) {
    return obj[name];
  };
}

it('should be able to put and read the queue', 3, function(t, db) {
  var q = queue(db);
  var data = {
    id: 1,
    name: 'Eugene',
    value: 42
  };
  q.queue.push(data, read);
  function read(err) {
    t.notOk(err);
    q.queue.read(function (err, value) {
      t.notOk(err);
      t.deepEquals(data, value);
    });
  }
});

it('reading should dequeue the item', 4, function(t, db) {
  var q = queue(db);
  var data = {
    id: 1,
    name: 'Eugene',
    value: 42
  };
  q.queue.push(data, read);
  function read(err) {
    t.notOk(err);
    q.queue.read(function (err, value, key) {
      t.notOk(err);
      t.deepEquals(data, value);
      get(key);
    });
  }

  function get(key) {
    q.get(key, check);
  }

  function check(err, value) {
    t.ok(err);
  }
});

it('reading should be atomic', 1, function(t, db) {
  var q = queue(db);
  var next = after(5, dequeue);
  range(0, 5).forEach(function (i) {
    q.queue.push({
      id: i,
      name: 'Name ' + i,
      value: 42 + i
    }, next);
  });

  var results = [];
  function dequeue() {
    var next = after(5, check);
    range(0, 5).forEach(function (i) {
      q.queue.read(function (err, value, key) {
        results.push(value);
        next();
      });
    });
  }

  function check() {
    var hits = uniq(results.map(prop('id')));
    t.equal(hits.length, 5);
  }
});

it('reading should block until data is there', 3, function(t, db) {
  var q = queue(db);
  var data = {
    id: 1,
    name: 'Eugene',
    value: 42
  };
  q.queue.read(function (err, value) {
    t.notOk(err);
    t.deepEquals(data, value);
    t.ok(Date.now() - now >= 20);
  });
  var now = Date.now();
  setTimeout(function () {
    q.queue.push(data);
  }, 20);
});

it('should be able to change the prioritation strategy', 5, function(t, db) {
  var q = queue(db, prop('value'));
  var next = after(5, dequeue);
  range(0, 5).forEach(function (i) {
    q.queue.push({
      id: i,
      name: 'Name ' + i,
      value: -(42 + i)
    }, next);
  });

  var results = [];
  function dequeue() {
    var next = after(5, check);
    range(0, 5).forEach(function (i) {
      q.queue.read(function (err, value, key) {
        results.push(value);
        next();
      });
    });
  }

  function check() {
    var last = -Infinity;
    results.forEach(function (result) {
      t.ok(result.value > last);
      last = result.value;
    });
  }
});
{'fraction_non_alphanumeric': 0.1875, 'fraction_numerical': 0.011659021406727829, 'mean_word_length': 2.493324432576769, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 6, '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': '5235466', 'n_tokens_mistral': 2010, 'n_tokens_neox': 1944, 'n_words': 467}