feat: Use pull net

This commit is contained in:
Maciej Krüger 2018-08-02 17:23:27 +02:00
parent a0c23e49f7
commit dc38263ec4
No known key found for this signature in database
GPG Key ID: 0D948CE19CF49C5F
4 changed files with 20 additions and 69 deletions

View File

@ -51,6 +51,7 @@
"mafmt": "^6.0.0", "mafmt": "^6.0.0",
"multiaddr": "^4.0.0", "multiaddr": "^4.0.0",
"once": "^1.4.0", "once": "^1.4.0",
"pull-net": "github:mkg20001/pull-net",
"stream-to-pull-stream": "^1.7.2" "stream-to-pull-stream": "^1.7.2"
}, },
"contributors": [ "contributors": [

View File

@ -9,22 +9,22 @@ module.exports = (socket) => {
let ma let ma
try { try {
if (socket.remoteFamily === 'IPv6') { if (socket.remoteAddress.family === 'IPv6') {
const addr = new Address6(socket.remoteAddress) const addr = new Address6(socket.remoteAddress.address)
if (addr.v4) { if (addr.v4) {
const ip4 = addr.to4().correctForm() const ip4 = addr.to4().correctForm()
ma = multiaddr('/ip4/' + ip4 + ma = multiaddr('/ip4/' + ip4 +
'/tcp/' + socket.remotePort '/tcp/' + socket.remoteAddress.port
) )
} else { } else {
ma = multiaddr('/ip6/' + socket.remoteAddress + ma = multiaddr('/ip6/' + socket.remoteAddress.address +
'/tcp/' + socket.remotePort '/tcp/' + socket.remoteAddress.port
) )
} }
} else { } else {
ma = multiaddr('/ip4/' + socket.remoteAddress + ma = multiaddr('/ip4/' + socket.remoteAddress.address +
'/tcp/' + socket.remotePort) '/tcp/' + socket.remoteAddress.port)
} }
} catch (err) { } catch (err) {
log(err) log(err)

View File

@ -1,7 +1,6 @@
'use strict' 'use strict'
const net = require('net') const {connect} = require('pull-net')
const toPull = require('stream-to-pull-stream')
const mafmt = require('mafmt') const mafmt = require('mafmt')
const withIs = require('class-is') const withIs = require('class-is')
const includes = require('lodash.includes') const includes = require('lodash.includes')
@ -27,23 +26,7 @@ class TCP {
const cOpts = ma.toOptions() const cOpts = ma.toOptions()
log('Connecting to %s %s', cOpts.port, cOpts.host) log('Connecting to %s %s', cOpts.port, cOpts.host)
const rawSocket = net.connect(cOpts) const conn = new Connection(connect(cOpts.port, cOpts.host, callback))
rawSocket.once('timeout', () => {
log('timeout')
rawSocket.emit('error', new Error('Timeout'))
})
rawSocket.once('error', callback)
rawSocket.once('connect', () => {
rawSocket.removeListener('error', callback)
callback()
})
const socket = toPull.duplex(rawSocket)
const conn = new Connection(socket)
conn.getObservedAddrs = (callback) => { conn.getObservedAddrs = (callback) => {
return callback(null, [ma]) return callback(null, [ma])

View File

@ -4,29 +4,23 @@ const multiaddr = require('multiaddr')
const Connection = require('interface-connection').Connection const Connection = require('interface-connection').Connection
const os = require('os') const os = require('os')
const includes = require('lodash.includes') const includes = require('lodash.includes')
const net = require('net') const {createServer} = require('pull-net')
const toPull = require('stream-to-pull-stream')
const EventEmitter = require('events').EventEmitter const EventEmitter = require('events').EventEmitter
const debug = require('debug') const debug = require('debug')
const log = debug('libp2p:tcp:listen') const log = debug('libp2p:tcp:listen')
const getMultiaddr = require('./get-multiaddr') const getMultiaddr = require('./get-multiaddr')
const IPFS_CODE = 421 const IPFS_CODE = 421
const CLOSE_TIMEOUT = 2000
function noop () {} function noop () {}
module.exports = (handler) => { module.exports = (handler) => {
const listener = new EventEmitter() const listener = new EventEmitter()
const server = net.createServer((socket) => { const server = createServer((stream) => {
// Avoid uncaught errors cause by unstable connections const addr = getMultiaddr(stream)
socket.on('error', noop)
const addr = getMultiaddr(socket)
if (!addr) { if (!addr) {
if (socket.remoteAddress === undefined) { if (stream.remoteAddress === undefined || stream.remoteAddress.address === 'undefined') {
log('connection closed before p2p connection made') log('connection closed before p2p connection made')
} else { } else {
log('error interpreting incoming p2p connection') log('error interpreting incoming p2p connection')
@ -36,25 +30,16 @@ module.exports = (handler) => {
log('new connection', addr.toString()) log('new connection', addr.toString())
const s = toPull.duplex(socket) stream.getObservedAddrs = (cb) => {
s.getObservedAddrs = (cb) => {
cb(null, [addr]) cb(null, [addr])
} }
trackSocket(server, socket) const conn = new Connection(stream)
const conn = new Connection(s)
handler(conn) handler(conn)
listener.emit('connection', conn) listener.emit('connection', conn)
}) })
server.on('listening', () => listener.emit('listening')) listener.emit('listening')
server.on('error', (err) => listener.emit('error', err))
server.on('close', () => listener.emit('close'))
// Keep track of open connections to destroy in case of timeout
server.__connections = {}
listener.close = (options, callback) => { listener.close = (options, callback) => {
if (typeof options === 'function') { if (typeof options === 'function') {
@ -64,18 +49,9 @@ module.exports = (handler) => {
callback = callback || noop callback = callback || noop
options = options || {} options = options || {}
const timeout = setTimeout(() => { server.close((err, ...a) => {
log('unable to close graciously, destroying conns') listener.emit('close')
Object.keys(server.__connections).forEach((key) => { callback(err, ...a)
log('destroying %s', key)
server.__connections[key].destroy()
})
}, options.timeout || CLOSE_TIMEOUT)
server.close(callback)
server.once('close', () => {
clearTimeout(timeout)
}) })
} }
@ -145,12 +121,3 @@ function getIpfsId (ma) {
return tuple[0] === IPFS_CODE return tuple[0] === IPFS_CODE
})[0][1] })[0][1]
} }
function trackSocket (server, socket) {
const key = `${socket.remoteAddress}:${socket.remotePort}`
server.__connections[key] = socket
socket.on('close', () => {
delete server.__connections[key]
})
}