<commit_before>/*
 * queue.js
 */
var util = require('util');
var MongoClient = require('mongodb').MongoClient;

var logger, client;


function MongoInserter(collection){
  this._coll = collection;
}
MongoInserter.prototype.start = function(items, callback){
  this._items = items;
  this._callback = callback;
  this._insertNext();
  this._insertedCount = 0; 
}
MongoInserter.prototype._insertNext = function(){
  var item = this._items.shift();
  if (item == undefined) {
    this._callback(null, this._insertedCount);
    return;
  }

  if (!("key" in item)) {
    logger.warn("Ignored item without 'key'.");
    this._insertNext();
    return;
  }

  if (!("priority" in item)) {
    item.priority = 10;
  }

  logger.debug("db." + this._coll.collectionName + ".upsert: " + util.inspect(item));

  var me = this;

  var updateSet = { 
      $inc : { priority: item.priority }
    };

  if ("value" in item) {
    updateSet["$set"] = { value: item.value };
  }

  me._coll.update({ "key" : item.key }, updateSet,
      { safe: true, upsert:true },
      function(err) {
        if (err) {
          me._callback(err, me._insertedCount);
          return;
        }

        me._insertedCount++;
        me._insertNext();
      });
}


exports.initialize = function(dbUri, param, mylogger, callback){
  logger = mylogger;
  callback = callback || (function(){});

  var params = param || {};
  if (!("auto_reconnect" in params))
     params["auto_reconnect"] = true;

  suffix = params.suffix || "";
  delete params.suffix;

  MongoClient.connect(dbUri, params, function(err, db){
      if (err) {
        callback(err);
        return;
      }
      client = db;
      
      callback(null);
    });
};


exports.terminate = function(){
  if (client) {
    client.close();
  }
}


exports.dataIn = function(target, items, callback){
  client.collection(target + suffix, function(err, coll) {
      if (err) {
        callback(err);
        return;
      }

      new MongoInserter(coll).start(items, callback);
    });
};


exports.dataOut = function(target, callback){
  var cb = function(err, result){
      result = result || null;
      return callback(err, result);
  };

  client.collection(target + suffix, function(err, coll) {
      if (err) {
        cb(err);
        return;
      }

      var sort = [ ["priority", -1], ["_id", 1] ];

      coll.find({}, {'limit': 1, 'sort': sort },
        function(err, cursor){
          if (err) {
            cb(err);
            return;
          }

          cursor.toArray(function(err, docs){
              if (!docs || docs.length == 0) {
                cb(null, null);
                return;
              }

              var doc = docs[0];
              var docId = doc._id;
              logger.debug("Data-out: " + util.inspect(doc));

              delete doc["_id"];

              if ( cb(null, doc) ) {
                coll.remove({ "_id": docId }, function(err, result){
                  logger.debug("Removed document _id:" + docId);
                });
              } else {
                logger.error("Client not received document _id:" + docId);
              }
            });
        });
    });
};


exports.clear = function(target, callback){
  client.collection(target + suffix, function(err, coll) {
      if (err) {
        callback(err);
        return;
      }

      coll.remove({}, function(err, result){
          callback(err);
        });
    });
};

<commit_msg>Fix to create mongodb client<commit_after>/*
 * queue.js
 */
const util = require('util')
const MongoClient = require('mongodb').MongoClient

let logger, client, db;


function MongoInserter(collection){
  this._coll = collection;
}
MongoInserter.prototype.start = function(items, callback){
  this._items = items;
  this._callback = callback;
  this._insertNext();
  this._insertedCount = 0; 
}
MongoInserter.prototype._insertNext = function(){
  var item = this._items.shift();
  if (item == undefined) {
    this._callback(null, this._insertedCount);
    return;
  }

  if (!("key" in item)) {
    logger.warn("Ignored item without 'key'.");
    this._insertNext();
    return;
  }

  if (!("priority" in item)) {
    item.priority = 10;
  }

  logger.debug("db." + this._coll.collectionName + ".upsert: " + util.inspect(item));

  var me = this;

  var updateSet = { 
      $inc : { priority: item.priority }
    };

  if ("value" in item) {
    updateSet["$set"] = { value: item.value };
  }

  me._coll.update({ "key" : item.key }, updateSet,
      { safe: true, upsert:true },
      function(err) {
        if (err) {
          me._callback(err, me._insertedCount);
          return;
        }

        me._insertedCount++;
        me._insertNext();
      });
}


exports.initialize = (dbUri, param, mylogger, callback) => {
  logger = mylogger;
  callback = callback || (function(){});

  const params = param || {};
  if (!("auto_reconnect" in params))
     params["auto_reconnect"] = true;

  const { suffix: suffix_, ...otherParams } = params
  suffix = suffix_ || ''

  const lastSlashPost = dbUri.lastIndexOf('/')
  const dbEndpoint = dbUri.substr(0, lastSlashPost)
  const dbName = dbUri.substr(lastSlashPost + 1)

  MongoClient.connect(dbEndpoint, otherParams, (err, client_) => {
    if (err) {
      callback(err)
      return
    }
    client = client_
    db = client_.db(dbName)
    callback(null)
  })
}


exports.terminate = function(){
  if (client) {
    client.close();
  }
}


exports.dataIn = function(target, items, callback){
  db.collection(target + suffix, function(err, coll) {
      if (err) {
        callback(err);
        return;
      }

      new MongoInserter(coll).start(items, callback);
    });
};


exports.dataOut = function(target, callback){
  var cb = function(err, result){
      result = result || null;
      return callback(err, result);
  };

  db.collection(target + suffix, function(err, coll) {
      if (err) {
        cb(err);
        return;
      }

      var sort = [ ["priority", -1], ["_id", 1] ];

      coll.find({}, {'limit': 1, 'sort': sort },
        function(err, cursor){
          if (err) {
            cb(err);
            return;
          }

          cursor.toArray(function(err, docs){
              if (!docs || docs.length == 0) {
                cb(null, null);
                return;
              }

              var doc = docs[0];
              var docId = doc._id;
              logger.debug("Data-out: " + util.inspect(doc));

              delete doc["_id"];

              if ( cb(null, doc) ) {
                coll.remove({ "_id": docId }, function(err, result){
                  logger.debug("Removed document _id:" + docId);
                });
              } else {
                logger.error("Client not received document _id:" + docId);
              }
            });
        });
    });
};


exports.clear = function(target, callback){
  db.collection(target + suffix, function(err, coll) {
      if (err) {
        callback(err);
        return;
      }

      coll.remove({}, function(err, result){
          callback(err);
        });
    });
};


{'fraction_non_alphanumeric': 0.1561840439003799, 'fraction_numerical': 0.002532714225411566, 'mean_word_length': 2.0877497827975673, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, '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': '6833016', 'n_tokens_mistral': 2261, 'n_tokens_neox': 2162, 'n_words': 483}