<commit_before>var KBucket = require('k-bucket')
var ms = require('ms')
var url = require('url')
var paramap = require('pull-paramap')
var asyncInterval = require('asyncinterval')
var valid = require("muxrpc-validation")()
var Manifest = require("./manifest")("./dht.md", __filename)

function arbiter(incumbent, candidate) {
  if (incumbent.lastSeen > candidate.lastSeen) {
    return incumbent
  }

  return candidate
}

var path = require('path')

module.exports = function (api,_) {
  var notify = _.notify()
  var source = notify.listen()
  source.path = 'dht'  
  source.manifest = Manifest.manifest()
  source.usage = valid.async(Manifest.usage,["object?"])
  
  var config = api.config.dht || {}

  config.bucketSize = config.bucketSize || 8
  config.alpha = config.alpha || 3
  config.bucketRefresh = config.bucketRefresh || ms('60m')
  config.replicate = config.replicate || ms('60m')

  config.pingInterval = config.pingInterval || ms('15s')
  api.config.bootstrap = api.config.bootstrap || []
  if (api.config.bootstrap && api.config.bootstrap.length === 0) api.logger.warn('no bootstrap url configured')


  var bucket = new KBucket({ localNodeId: api.config.info.id, numberOfNodesPerKBucket: config.bucketSize, dontSplit: true, arbiter: arbiter })

  var addrs = []

  function connect(addresses, cb) {
    var retry = 3
    var _connect = function (addresses, cb) {
      retry--
      var addrs = addresses.slice(0).filter(function (addr) {
        return api.swarm.protoNames().indexOf(url.parse(addr).protocol) !== -1
      }).sort(function sortFunc(a, b) {
        var sortingArr = api.swarm.protoNames()
        return sortingArr.indexOf(url.parse(a[1]).protocol) - sortingArr.indexOf(url.parse(b[1]).protocol)
      })
      api.swarm.connect(addrs.shift(), {}, function cb2(err, con) {
        if (addrs.length > 0 && err != null) return api.swarm.connect(addrs.shift(), cb2)
        if (err != null && retry > 0) return _connect(addresses, cb)
        cb(err, con)
      })
    }
    _connect(addresses, cb)
  }

  function connectToClosest(map) {
    return _(source.findNode(api.id), _.filterNot(function (contact) {
      return api.id == contact.id
    }), paramap(function (contact, cb) {
      connect(contact.addrs, function (err, connection) {
        map(err, connection, cb)
      })
    }, config.alpha))
  }


  function bootstrap(cb) {

    var closest = source.findNode(api.id)

    if (closest.length === 0 && api.config.bootstrap.length > 0) {
      api.logger.debug('starting bootstraping on', api.id)

      connect(api.config.bootstrap, function (err, conn) {
        if (err) {
          api.logger.warn('unable to connect to bootstrap peer', err)
          if (cb) return cb()
          return
        }
        if (conn)
          ping(conn, function () {
            if (cb) cb(conn)
            api.logger.debug('bootstrapping on', api.id, 'is done')

          })
        else cb()
      })
      return
    }
    
    if (cb) cb()
  }


  var replicate = function (cb) {

    _(connectToClosest(function (err, connection, cb) {
      if (err) return cb()
      
      api.logger.info('replicating data from peer', connection.peerID)

      api.db.load(connection.peer.db, function (err) {
          if(connection.peer.db.read)return connection.peer.db.read(function(_err){
            api.logger.info('data from', connection.peerID, 'replicated')  
            if (cb) cb(err||_err)
          })

          api.logger.info('data from', connection.peerID, 'replicated')     
          if (cb) cb(err)          
      })

    }
    ), _.onEnd(function () {
      if (cb) cb()
    }))

  }

  var timers = {}
  timers.bootstrap = asyncInterval(function (next) { 
    if(timers.bootstrap){
      if (bucket.count() < 1) return bootstrap(function (conn) {
        next()
      })
      next()
   }}, ms('40s'))
  timers.ping = {}
  
  timers.refresh = asyncInterval(function (next) { 
    if(timers.refresh)refresh(next) }, config.bucketRefresh)
  timers.replicate = asyncInterval(function (next) { if(timers.replicate)replicate(next) }, config.replicate)

  function ping(e, cb) {
    api.logger.debug('ping peer ', e.peerID)
    e.peer.dht.ping(function (err, addr) {
      api.logger.debug('pinged peer ', e.peerID)<commit_msg>connect if no connection<commit_after>var KBucket = require('k-bucket')
var ms = require('ms')
var url = require('url')
var paramap = require('pull-paramap')
var asyncInterval = require('asyncinterval')
var valid = require("muxrpc-validation")()
var Manifest = require("./manifest")("./dht.md", __filename)

function arbiter(incumbent, candidate) {
  if (incumbent.lastSeen > candidate.lastSeen) {
    return incumbent
  }

  return candidate
}

var path = require('path')

module.exports = function (api,_) {
  var notify = _.notify()
  var source = notify.listen()
  source.path = 'dht'  
  source.manifest = Manifest.manifest()
  source.usage = valid.async(Manifest.usage,["object?"])
  
  var config = api.config.dht || {}

  config.bucketSize = config.bucketSize || 8
  config.alpha = config.alpha || 3
  config.bucketRefresh = config.bucketRefresh || ms('60m')
  config.replicate = config.replicate || ms('60m')

  config.pingInterval = config.pingInterval || ms('15s')
  api.config.bootstrap = api.config.bootstrap || []
  if (api.config.bootstrap && api.config.bootstrap.length === 0) api.logger.warn('no bootstrap url configured')


  var bucket = new KBucket({ localNodeId: api.config.info.id, numberOfNodesPerKBucket: config.bucketSize, dontSplit: true, arbiter: arbiter })

  var addrs = []

  function connect(addresses, cb) {
    var retry = 3
    var _connect = function (addresses, cb) {
      retry--
      var addrs = addresses.slice(0).filter(function (addr) {
        return api.swarm.protoNames().indexOf(url.parse(addr).protocol) !== -1
      }).sort(function sortFunc(a, b) {
        var sortingArr = api.swarm.protoNames()
        return sortingArr.indexOf(url.parse(a[1]).protocol) - sortingArr.indexOf(url.parse(b[1]).protocol)
      })
      api.swarm.connect(addrs.shift(), {}, function cb2(err, con) {
        if (addrs.length > 0 && err != null) return api.swarm.connect(addrs.shift(), cb2)
        if (err != null && retry > 0) return _connect(addresses, cb)
        cb(err, con)
      })
    }
    _connect(addresses, cb)
  }

    function connectToClosest(map) {
      return _(source.findNode(api.id), _.filterNot(function (contact) {
        return api.id == contact.id
      }), paramap(function (contact, cb) {
        connect(contact.addrs, function (err, connection) {
          map(err, connection, cb)
        })
      }, config.alpha))
    }


  function bootstrap(cb) {

    var closest = source.findNode(api.id)

    if (closest.length === 0 && api.config.bootstrap.length > 0) {
      api.logger.debug('starting bootstraping on', api.id)

      connect(api.config.bootstrap, function (err, conn) {
        if (err) {
          api.logger.warn('unable to connect to bootstrap peer', err)
          if (cb) return cb()
          return
        }
        if (conn)
          ping(conn, function () {
            if (cb) cb(conn)
            api.logger.debug('bootstrapping on', api.id, 'is done')

          })
        else cb()
      })
      return
    }
    
    if (cb) cb()
  }


  var replicate = function (cb) {

    _(connectToClosest(function (err, connection, cb) {
      if (err) return cb()
      
      api.logger.info('replicating data from peer', connection.peerID)

      api.db.load(connection.peer.db, function (err) {
          if(connection.peer.db.read)return connection.peer.db.read(function(_err){
            api.logger.info('data from', connection.peerID, 'replicated')  
            if (cb) cb(err||_err)
          })

          api.logger.info('data from', connection.peerID, 'replicated')     
          if (cb) cb(err)          
      })

    }
    ), _.onEnd(function () {
      if (cb) cb()
    }))

  }


  var timers = {}
  timers.bootstrap = asyncInterval(function (next) { 
    if(timers.bootstrap){
      if (bucket.count() < 1) return bootstrap(function (conn) {
        next()
      })
      
      if(Object.keys(api.swarm._connections).length == 0 )
      return _(connectToClosest(function (err, connection, cb) {
        if (err) return cb()
        
        cb()
    
      }
      ), _.onEnd(next))

      next()
   }}, ms('1m'))

  timers.ping = {}
  
  timers.refresh = asyncInterval(function (next) { 
    if(timers.refresh)refresh(next) }, config.bucketRefresh)
  timers.replicate = asyncInterval(function (next) { if(timers.replicate)replicate(next) }, config.replicate)

  function ping(e, cb) {
    api.logger.debug('ping peer ', e.peerID)
    e.peer.dht.ping(function (err, addr) {
      api.logger.debug('pinged peer ', e.peerID)
{'fraction_non_alphanumeric': 0.14912480109115708, 'fraction_numerical': 0.005228461013866788, 'mean_word_length': 2.7331353415358506, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '6043442', 'n_tokens_mistral': 2887, 'n_tokens_neox': 2763, 'n_words': 599}