<commit_before>var tcp = require('net')
var Select = require('multistream-select').Select
var Interactive = require('multistream-select').Interactive
var spdy = require('spdy-transport')
var log = require('ipfs-logger').group('swarm')
var async = require('async')
var EventEmitter = require('events').EventEmitter
var util = require('util')

exports = module.exports = Swarm

util.inherits(Swarm, EventEmitter)

function Swarm () {
  var self = this

  if (!(self instanceof Swarm)) {
    throw new Error('Swarm must be called with new')
  }

  self.port = parseInt(process.env.IPFS_SWARM_PORT, 10) || 4001

  self.connections = {}
  self.handles = []

  // set the listener

  self.listen = function (port, ready) {
    if (!ready) {
      ready = function noop () {}
    }
    if (typeof port === 'function') {
      ready = port
    } else if (port) {
      self.port = port
    }

    tcp.createServer(function (socket) {
      var ms = new Select()
      ms.handle(socket)
      ms.addHandler('/spdy/3.1.0', function (ds) {
        log.info('Negotiated spdy with incoming socket')

        var conn = spdy.connection.create(ds, {
          protocol: 'spdy',
          isServer: true
        })

        conn.start(3.1)

        self.emit('connection', conn)

        // attach multistream handlers to incoming streams
        conn.on('stream', registerHandles)

        // IDENTIFY DOES THAT FOR US
        // conn.on('close', function () { delete self.connections[conn.peerId] })
      })
    }).listen(self.port, ready)
  }

  // interface

  // open stream account for connection reuse
  self.openStream = function (peer, protocol, cb) {

    // If no connection open yet, open it
    if (!self.connections[peer.id.toB58String()]) {

      // Establish a socket with one of the addresses
      var socket
      async.eachSeries(peer.multiaddrs, function (multiaddr, next) {
        if (socket) { return next() }

        var tmp = tcp.connect(multiaddr.toOptions(), function () {
          socket = tmp
          next()
        })

        tmp.once('error', function (err) {
          log.warn(multiaddr.toString(), 'on', peer.id.toB58String(), 'not available', err)
          next()
        })

      }, function done () {
        if (!socket) {
          return cb(new Error('Not able to open a scoket with peer - ',
                peer.id.toB58String()))
        }
        gotSocket(socket)
      })
    } else {
      createStream(peer, protocol, cb)
    }

    // do the spdy people dance (multistream-select into spdy)
    function gotSocket (socket) {
      var msi = new Interactive()
      msi.handle(socket, function () {
        msi.select('/spdy/3.1.0', function (err, ds) {
          if (err) { cb(err) }

          var conn = spdy.connection.create(ds, { protocol: 'spdy', isServer: false })
          conn.start(3.1)
          conn.on('stream', registerHandles)
          self.connections[peer.id.toB58String()] = conn

          conn.on('close', function () { delete self.connections[peer.id.toB58String()] })

          createStream(peer, protocol, cb)<commit_msg>clean the code a bit<commit_after>var tcp = require('net')
var Select = require('multistream-select').Select
var Interactive = require('multistream-select').Interactive
var spdy = require('spdy-transport')
var log = require('ipfs-logger').group('swarm')
var async = require('async')
var EventEmitter = require('events').EventEmitter
var util = require('util')

exports = module.exports = Swarm

util.inherits(Swarm, EventEmitter)

function Swarm () {
  var self = this

  if (!(self instanceof Swarm)) {
    throw new Error('Swarm must be called with new')
  }

  self.port = parseInt(process.env.IPFS_SWARM_PORT, 10) || 4001

  self.connections = {}
  self.handles = []

  // set the listener

  self.listen = function (port, ready) {
    if (!ready) {
      ready = function noop () {}
    }
    if (typeof port === 'function') {
      ready = port
    } else if (port) {
      self.port = port
    }

    tcp.createServer(function (socket) {
      var ms = new Select()
      ms.handle(socket)
      ms.addHandler('/spdy/3.1.0', function (ds) {
        log.info('Negotiated spdy with incoming socket')

        var conn = spdy.connection.create(ds, {
          protocol: 'spdy',
          isServer: true
        })

        conn.start(3.1)

        self.emit('connection', conn)

        // attach multistream handlers to incoming streams
        conn.on('stream', registerHandles)

      // IDENTIFY DOES THAT FOR US
      // conn.on('close', function () { delete self.connections[conn.peerId] })
      })
    }).listen(self.port, ready)
  }

  // interface

  // open stream account for connection reuse
  self.openStream = function (peer, protocol, cb) {
    // If no connection open yet, open it
    if (!self.connections[peer.id.toB58String()]) {
      // Establish a socket with one of the addresses
      var socket
      async.eachSeries(peer.multiaddrs, function (multiaddr, next) {
        if (socket) { return next() }

        var tmp = tcp.connect(multiaddr.toOptions(), function () {
          socket = tmp
          next()
        })

        tmp.once('error', function (err) {
          log.warn(multiaddr.toString(), 'on', peer.id.toB58String(), 'not available', err)
          next()
        })

      }, function done () {
        if (!socket) {
          return cb(new Error('Not able to open a scoket with peer - ',
            peer.id.toB58String()))
        }
        gotSocket(socket)
      })
    } else {
      createStream(peer, protocol, cb)
    }

    // do the spdy people dance (multistream-select into spdy)
    function gotSocket (socket) {
      var msi = new Interactive()
      msi.handle(socket, function () {
        msi.select('/spdy/3.1.0', function (err, ds) {
          if (err) { cb(err) }

          var conn = spdy.connection.create(ds, { protocol: 'spdy', isServer: false })
          conn.start(3.1)
          conn.on('stream', registerHandles)
          self.connections[peer.id.toB58String()] = conn

          conn.on('close', function () { delete self.connections[peer.id.toB58String()] })

          createStream(peer, protocol, cb)
{'fraction_non_alphanumeric': 0.13663590739841347, 'fraction_numerical': 0.008418326048243484, 'mean_word_length': 2.3889193636862314, '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': '1719502', 'n_tokens_mistral': 1912, 'n_tokens_neox': 1839, 'n_words': 479}