<commit_before>// relyq
// A reliable task queue

// vendor
var async = require('async'),
  _ = require('underscore'),
  redisPkg = require('redis'),
  simpleq = require('simpleq');

// local
var DeferredTaskList = require('./deferred');

// -- Master Type: Q --
// The master type, a task queue
function Q(redis, preopts) {
  // handle forgetting a 'new'
  if (!(this instanceof Q)) {
    return new Q(redis, preopts);
  }

  var Constructor = this.constructor;
  this.clone = this.clone || function () {
    return new Constructor(redisPkg.createClient(redis.port, redis.host, redis.options), preopts);
  };

  this._delimeter = preopts.delimeter || ':';

  var idfield = this._idfield = preopts.idfield || 'id';
  this._getid = preopts.getid || function (task) {
    return task[idfield];
  };

  this._prefix = preopts.prefix || preopts;
  this._clean_finish = preopts.clean_finish === undefined || preopts.clean_finish;
  this._keep_storage = preopts.clean_finish === 'keep_storage';

  this.todo = new simpleq.Q(redis, this._prefix + this._delimeter + 'todo');
  this.doing = new simpleq.Q(redis, this._prefix + this._delimeter + 'doing');
  this.failed = new simpleq.Q(redis, this._prefix + this._delimeter + 'failed');
  if (!this._clean_finish) {
    this.done = new simpleq.Q(redis, this._prefix + this._delimeter + 'done');
  }

  if (preopts.allow_defer) {
    this.deferred = new DeferredTaskList(this, {
      interval: preopts.defer_interval,
      key: this._prefix + this._delimeter + 'deferred',
      redis: redis,
    });
  }
}

// @overridable
// Get a task object from its reference id
Q.prototype.get = function get(taskref, callback) {
  callback(null, taskref);
};

// @overridable
// Set a task object and return its reference ID
Q.prototype.set = function set(taskobj, taskref, callback) {
  callback();
};

// @overridable
// Delete the task obj and return its reference ID
Q.prototype.del = function del(taskobj, taskref, callback) {
  callback();
};

// Override these to add a new serialization
// refs get put into the queue
// @overridable
Q.prototype.ref = function (obj) {
  return obj;
};

// End the listeners that might be listening
Q.prototype.end = function (callback) {
  if (this.deferred) {
    this.deferred.end();
  }
  if (this._sql) {
    this._sql.on('end', callback).end();
  } else {
    callback();
  }
}

// -- Superclass methods ---

Q.prototype.push = function push(task, callback) {
  var ref = this.ref(task);
  async.parallel([
    _.bind(this.set, this, task, ref),
    _.bind(this.todo.push, this.todo, ref)
  ], function (err, results) {
    callback(err, results && results.length === 2 && results[1]);
  });
};

Q.prototype.defer = function defer(task, when, callback) {
  if (!this.deferred) {
    throw new Error('Must use option allow_defer to allow defer calls.');
  }

  var ref = this.ref(task);
  async.parallel([
    _.bind(this.set, this, task, ref),
    _.bind(this.deferred.defer, this.deferred, ref, when),
  ], function (err, results) {<commit_msg>Only call end callback once<commit_after>// relyq
// A reliable task queue

// vendor
var async = require('async'),
  _ = require('underscore'),
  redisPkg = require('redis'),
  simpleq = require('simpleq');

// local
var DeferredTaskList = require('./deferred');

// -- Master Type: Q --
// The master type, a task queue
function Q(redis, preopts) {
  // handle forgetting a 'new'
  if (!(this instanceof Q)) {
    return new Q(redis, preopts);
  }

  var Constructor = this.constructor;
  this.clone = this.clone || function () {
    return new Constructor(redisPkg.createClient(redis.port, redis.host, redis.options), preopts);
  };

  this._delimeter = preopts.delimeter || ':';

  var idfield = this._idfield = preopts.idfield || 'id';
  this._getid = preopts.getid || function (task) {
    return task[idfield];
  };

  this._prefix = preopts.prefix || preopts;
  this._clean_finish = preopts.clean_finish === undefined || preopts.clean_finish;
  this._keep_storage = preopts.clean_finish === 'keep_storage';

  this.todo = new simpleq.Q(redis, this._prefix + this._delimeter + 'todo');
  this.doing = new simpleq.Q(redis, this._prefix + this._delimeter + 'doing');
  this.failed = new simpleq.Q(redis, this._prefix + this._delimeter + 'failed');
  if (!this._clean_finish) {
    this.done = new simpleq.Q(redis, this._prefix + this._delimeter + 'done');
  }

  if (preopts.allow_defer) {
    this.deferred = new DeferredTaskList(this, {
      interval: preopts.defer_interval,
      key: this._prefix + this._delimeter + 'deferred',
      redis: redis,
    });
  }
}

// @overridable
// Get a task object from its reference id
Q.prototype.get = function get(taskref, callback) {
  callback(null, taskref);
};

// @overridable
// Set a task object and return its reference ID
Q.prototype.set = function set(taskobj, taskref, callback) {
  callback();
};

// @overridable
// Delete the task obj and return its reference ID
Q.prototype.del = function del(taskobj, taskref, callback) {
  callback();
};

// Override these to add a new serialization
// refs get put into the queue
// @overridable
Q.prototype.ref = function (obj) {
  return obj;
};

// End the listeners that might be listening
Q.prototype.end = function (callback) {
  if (this.deferred) {
    this.deferred.end();
  }
  if (this._sql) {
    this._sql.once('end', callback).end();
  } else {
    callback();
  }
}

// -- Superclass methods ---

Q.prototype.push = function push(task, callback) {
  var ref = this.ref(task);
  async.parallel([
    _.bind(this.set, this, task, ref),
    _.bind(this.todo.push, this.todo, ref)
  ], function (err, results) {
    callback(err, results && results.length === 2 && results[1]);
  });
};

Q.prototype.defer = function defer(task, when, callback) {
  if (!this.deferred) {
    throw new Error('Must use option allow_defer to allow defer calls.');
  }

  var ref = this.ref(task);
  async.parallel([
    _.bind(this.set, this, task, ref),
    _.bind(this.deferred.defer, this.deferred, ref, when),
  ], function (err, results) {
{'fraction_non_alphanumeric': 0.15275486638073243, 'fraction_numerical': 0.0006598482349059716, 'mean_word_length': 3.454812637766348, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5643469', 'n_tokens_mistral': 2017, 'n_tokens_neox': 1988, 'n_words': 522}