<commit_before>"use strict";
var stream      = require('stream');
var async       = require('async');
var util        = require('util');
var co          = require('co');
var _           = require('underscore');
var Q           = require('q');
var parsers     = require('./app/lib/streams/parsers/doc');
var constants   = require('./app/lib/constants');
var fileDAL     = require('./app/lib/dal/fileDAL');
var jsonpckg    = require('./package.json');
var router      = require('./app/lib/streams/router');
var multicaster = require('./app/lib/streams/multicaster');
var base58      = require('./app/lib/base58');
var crypto      = require('./app/lib/crypto');
var Peer        = require('./app/lib/entity/peer');
var signature   = require('./app/lib/signature');
var common      = require('./app/lib/common');
var directory   = require('./app/lib/directory');
var INNER_WRITE = true;

// Add methods to String and Date
common.shim();

function Server (dbConf, overrideConf) {

  stream.Duplex.call(this, { objectMode: true });

  let home = directory.getHome(dbConf.name, dbConf.home);
  var logger = require('./app/lib/logger')('server');
  var that = this;
  var connectionPromise, servicesPromise;
  that.conf = null;
  that.dal = null;
  that.version = jsonpckg.version;

  var documentsMapping = {};

  // Unused, but made mandatory by Duplex interface
  this._read = function () {
  };

  this._write = function (obj, enc, writeDone, isInnerWrite) {
    that.submit(obj, isInnerWrite, function (err, res) {
      if (isInnerWrite) {
        writeDone(err, res);
      } else {
        writeDone();
      }
    });
  };

  this.connectDB = function (useDefaultConf) {
    // Connect only once
    return connectionPromise || (connectionPromise = that.connect(useDefaultConf));
  };

  this.initWithServices = function (done) {
    // Init services only once
    return servicesPromise || (servicesPromise = that.connectDB()
        .then(that.initServices)
        .then(function(err) {
          done && done(err);
          return that;
        })
        .catch(done));
  };

  this.submit = function (obj, isInnerWrite, done) {
    return co(function *() {
      if (!obj.documentType) {
        throw 'Document type not given';
      }
      var action = documentsMapping[obj.documentType];
      try {
        let res;
        if (typeof action == 'function') {
          // Handle the incoming object
          res = yield Q.nbind(action, this)(obj);
        } else {
          throw 'Unknown document type \'' + obj.documentType + '\'';
        }
        if (res) {
          // Only emit valid documents
          that.emit(obj.documentType, res);
          that.push(res);
        }
        isInnerWrite ? done(null, res) : done();
      } catch (err) {
        logger.debug(err);
        isInnerWrite ? done(err, null) : done();
      }
    });
  };

  this.submitP = (obj, isInnerWrite) => Q.nbind(this.submit, this)(obj, isInnerWrite);

  this.connect = function (useDefaultConf) {
    // Init connection
    if (that.dal) {
      return Q();
    }
    var dbType = dbConf && dbConf.memory ? fileDAL.memory : fileDAL.file;
    return dbType(home)
      .then(function(dal){
        that.dal = dal;
        return that.dal.init(overrideConf, useDefaultConf);
      })
      .then(function(conf){
        that.conf = conf;
        // Default values
        var defaultValues = {
          remoteipv6:         that.conf.ipv6,
          remoteport:         that.conf.port,
          cpu:                1,
          c:                  constants.CONTRACT.DEFAULT.C,
          dt:                 constants.CONTRACT.DEFAULT.DT,
          ud0:                constants.CONTRACT.DEFAULT.UD0,
          stepMax:            constants.CONTRACT.DEFAULT.STEPMAX,
          sigDelay:           constants.CONTRACT.DEFAULT.SIGDELAY,
          sigValidity:        constants.CONTRACT.DEFAULT.SIGVALIDITY,
          msValidity:         constants.CONTRACT.DEFAULT.MSVALIDITY,
          sigQty:             constants.CONTRACT.DEFAULT.SIGQTY,
          sigWoT:             constants.CONTRACT.DEFAULT.SIGWOT,
          percentRot:         constants.CONTRACT.DEFAULT.PERCENTROT,
          blocksRot:          constants.CONTRACT.DEFAULT.BLOCKSROT,
          powDelay:           constants.CONTRACT.DEFAULT.POWDELAY,
          avgGenTime:         constants.CONTRACT.DEFAULT.AVGGENTIME,
          dtDiffEval:         constants.CONTRACT.DEFAULT.DTDIFFEVAL,
          medianTimeBlocks:   constants.CONTRACT.DEFAULT.MEDIANTIMEBLOCKS,
          rootoffset:         0,
          forksize:           constants.BRANCHES.DEFAULT_WINDOW_SIZE
        };
        _.keys(defaultValues).forEach(function(key){
          if (that.conf[key] == undefined) {
            that.conf[key] = defaultValues[key];
          }
        });
      });
  };

  this.start = function () {
    return that.checkConfig()
      .then(function (){
        // Add signing & public key functions to PeeringService
        logger.info('Node version: ' + that.version);
        logger.info('Node pubkey: ' + that.PeeringService.pubkey);
        return Q.nfcall(that.initPeer);
      });
  };

  this.recomputeSelfPeer = function() {
    return Q.nbind(that.PeeringService.generateSelfPeer, that.PeeringService)(that.conf, 0);
  };

  this.initPeer = function (done) {
    async.waterfall([
      function (next){
        that.checkConfig().then(next).catch(next);
      },
      function (next){
        that.PeeringService.regularCrawlPeers(next);
      },
      function (next){
        logger.info('Storing self peer...');
        that.PeeringService.regularPeerSignal(next);
      },
      function(next) {
        that.PeeringService.regularTestPeers(next);
      },
      function (next){
        that.PeeringService.regularSyncBlock(next);
      },
      function (next){
        that.BlockchainService.regularCleanMemory(next);
      }
    ], done);
  };

  let shouldContinue = true;

  this.stopBlockComputation = function() {
    shouldContinue = false;
  };

  this.startBlockComputation = function() {
    return co(function *() {
      while (shouldContinue) {
        try {
          let block = yield that.BlockchainService.startGeneration();
          if (block && shouldContinue) {
            try {
              yield that.singleWritePromise(block);
            } catch (err) {
              logger.warn('Proof-of-work self-submission: %s', err.message || err);
            }
          }
        }
        catch (e) {
          logger.error(e);
          shouldContinue = true;
        }
      }
    });
  };<commit_msg>Fix #334 Blockchain was stuck with transactions<commit_after>"use strict";
var stream      = require('stream');
var async       = require('async');
var util        = require('util');
var co          = require('co');
var _           = require('underscore');
var Q           = require('q');
var parsers     = require('./app/lib/streams/parsers/doc');
var constants   = require('./app/lib/constants');
var fileDAL     = require('./app/lib/dal/fileDAL');
var jsonpckg    = require('./package.json');
var router      = require('./app/lib/streams/router');
var base58      = require('./app/lib/base58');
var crypto      = require('./app/lib/crypto');
var signature   = require('./app/lib/signature');
var common      = require('./app/lib/common');
var directory   = require('./app/lib/directory');
var http2raw    = require('./app/lib/streams/parsers/http2raw');
var dos2unix    = require('./app/lib/dos2unix');
var INNER_WRITE = true;

// Add methods to String and Date
common.shim();

function Server (dbConf, overrideConf) {

  stream.Duplex.call(this, { objectMode: true });

  let home = directory.getHome(dbConf.name, dbConf.home);
  var logger = require('./app/lib/logger')('server');
  var that = this;
  var connectionPromise, servicesPromise;
  that.conf = null;
  that.dal = null;
  that.version = jsonpckg.version;

  var documentsMapping = {};

  // Unused, but made mandatory by Duplex interface
  this._read = function () {
  };

  this._write = function (obj, enc, writeDone, isInnerWrite) {
    that.submit(obj, isInnerWrite, function (err, res) {
      if (isInnerWrite) {
        writeDone(err, res);
      } else {
        writeDone();
      }
    });
  };

  this.connectDB = function (useDefaultConf) {
    // Connect only once
    return connectionPromise || (connectionPromise = that.connect(useDefaultConf));
  };

  this.initWithServices = function (done) {
    // Init services only once
    return servicesPromise || (servicesPromise = that.connectDB()
        .then(that.initServices)
        .then(function(err) {
          done && done(err);
          return that;
        })
        .catch(done));
  };

  this.submit = function (obj, isInnerWrite, done) {
    return co(function *() {
      if (!obj.documentType) {
        throw 'Document type not given';
      }
      var action = documentsMapping[obj.documentType];
      try {
        let res;
        if (typeof action == 'function') {
          // Handle the incoming object
          res = yield Q.nbind(action, this)(obj);
        } else {
          throw 'Unknown document type \'' + obj.documentType + '\'';
        }
        if (res) {
          // Only emit valid documents
          that.emit(obj.documentType, res);
          that.push(res);
        }
        isInnerWrite ? done(null, res) : done();
      } catch (err) {
        logger.debug(err);
        isInnerWrite ? done(err, null) : done();
      }
    });
  };

  this.submitP = (obj, isInnerWrite) => Q.nbind(this.submit, this)(obj, isInnerWrite);

  this.connect = function (useDefaultConf) {
    // Init connection
    if (that.dal) {
      return Q();
    }
    var dbType = dbConf && dbConf.memory ? fileDAL.memory : fileDAL.file;
    return dbType(home)
      .then(function(dal){
        that.dal = dal;
        return that.dal.init(overrideConf, useDefaultConf);
      })
      .then(function(conf){
        that.conf = conf;
        // Default values
        var defaultValues = {
          remoteipv6:         that.conf.ipv6,
          remoteport:         that.conf.port,
          cpu:                1,
          c:                  constants.CONTRACT.DEFAULT.C,
          dt:                 constants.CONTRACT.DEFAULT.DT,
          ud0:                constants.CONTRACT.DEFAULT.UD0,
          stepMax:            constants.CONTRACT.DEFAULT.STEPMAX,
          sigDelay:           constants.CONTRACT.DEFAULT.SIGDELAY,
          sigValidity:        constants.CONTRACT.DEFAULT.SIGVALIDITY,
          msValidity:         constants.CONTRACT.DEFAULT.MSVALIDITY,
          sigQty:             constants.CONTRACT.DEFAULT.SIGQTY,
          sigWoT:             constants.CONTRACT.DEFAULT.SIGWOT,
          percentRot:         constants.CONTRACT.DEFAULT.PERCENTROT,
          blocksRot:          constants.CONTRACT.DEFAULT.BLOCKSROT,
          powDelay:           constants.CONTRACT.DEFAULT.POWDELAY,
          avgGenTime:         constants.CONTRACT.DEFAULT.AVGGENTIME,
          dtDiffEval:         constants.CONTRACT.DEFAULT.DTDIFFEVAL,
          medianTimeBlocks:   constants.CONTRACT.DEFAULT.MEDIANTIMEBLOCKS,
          rootoffset:         0,
          forksize:           constants.BRANCHES.DEFAULT_WINDOW_SIZE
        };
        _.keys(defaultValues).forEach(function(key){
          if (that.conf[key] == undefined) {
            that.conf[key] = defaultValues[key];
          }
        });
      });
  };

  this.start = function () {
    return that.checkConfig()
      .then(function (){
        // Add signing & public key functions to PeeringService
        logger.info('Node version: ' + that.version);
        logger.info('Node pubkey: ' + that.PeeringService.pubkey);
        return Q.nfcall(that.initPeer);
      });
  };

  this.recomputeSelfPeer = function() {
    return Q.nbind(that.PeeringService.generateSelfPeer, that.PeeringService)(that.conf, 0);
  };

  this.initPeer = function (done) {
    async.waterfall([
      function (next){
        that.checkConfig().then(next).catch(next);
      },
      function (next){
        that.PeeringService.regularCrawlPeers(next);
      },
      function (next){
        logger.info('Storing self peer...');
        that.PeeringService.regularPeerSignal(next);
      },
      function(next) {
        that.PeeringService.regularTestPeers(next);
      },
      function (next){
        that.PeeringService.regularSyncBlock(next);
      },
      function (next){
        that.BlockchainService.regularCleanMemory(next);
      }
    ], done);
  };

  let shouldContinue = true;

  this.stopBlockComputation = function() {
    shouldContinue = false;
  };

  this.startBlockComputation = function() {
    return co(function *() {
      while (shouldContinue) {
        try {
          let block = yield that.BlockchainService.startGeneration();
          if (block && shouldContinue) {
            try {
              let obj = parsers.parseBlock.syncWrite(dos2unix(block.getRawSigned()));
              yield that.singleWritePromise(obj);
            } catch (err) {
              logger.warn('Proof-of-work self-submission: %s', err.message || err);
            }
          }
        }
        catch (e) {
          logger.error(e);
          shouldContinue = true;
        }
      }
    });
  };
{'fraction_non_alphanumeric': 0.12988574864702346, 'fraction_numerical': 0.002254960914010824, 'mean_word_length': 2.0862908837856646, 'pattern_counts': {'":': 0, '<': 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': '5947398', 'n_tokens_mistral': 3911, 'n_tokens_neox': 3850, 'n_words': 757}