Compare commits

...

16 Commits

Author SHA1 Message Date
46672b669f chore: release version v0.10.1 2017-04-13 15:10:07 +01:00
31f69ec3c2 chore: update contributors 2017-04-13 15:10:07 +01:00
afed19439a chore: update deps 2017-04-10 14:32:05 -04:00
4b78faae3f Merge pull request #79 from libp2p/fix/catch-socket-reset-errors
fix: catch errors on incomming sockets
2017-04-10 14:19:24 -04:00
e204517a51 fix: catch errors on incomming sockets 2017-04-07 11:51:56 -04:00
4694f9dd47 chore: release version v0.10.0 2017-03-27 16:21:35 +01:00
dd6e067125 chore: update contributors 2017-03-27 16:21:34 +01:00
fb382a3e2a codestyle: fix linting 2017-03-27 16:21:26 +01:00
fd741367a2 chore: update deps 2017-03-27 16:19:30 +01:00
4d4f295dd5 fix(dial): proper error handling on dial (#77) 2017-03-27 16:18:55 +01:00
0edc487b23 chore: release version v0.9.4 2017-03-21 14:27:22 +00:00
c3ad8ce47a chore: update contributors 2017-03-21 14:27:22 +00:00
665b755922 chore: update aegir 2017-03-21 14:27:16 +00:00
1a550bac26 chore: release version v0.9.3 2017-02-09 09:01:06 -08:00
51c5594d49 chore: update contributors 2017-02-09 09:01:05 -08:00
d5476caee1 missing dep 2017-02-09 09:00:58 -08:00
4 changed files with 94 additions and 78 deletions

View File

@ -1,6 +1,6 @@
{ {
"name": "libp2p-tcp", "name": "libp2p-tcp",
"version": "0.9.2", "version": "0.10.1",
"description": "Node.js implementation of the TCP module that libp2p uses, which implements the interface-connection and interface-transport interfaces", "description": "Node.js implementation of the TCP module that libp2p uses, which implements the interface-connection and interface-transport interfaces",
"main": "src/index.js", "main": "src/index.js",
"scripts": { "scripts": {
@ -33,19 +33,22 @@
"node": ">=4.0.0" "node": ">=4.0.0"
}, },
"devDependencies": { "devDependencies": {
"interface-transport": "~0.3.3", "aegir": "^11.0.1",
"aegir": "^10.0.0",
"chai": "^3.5.0", "chai": "^3.5.0",
"dirty-chai": "^1.2.2",
"interface-transport": "~0.3.5",
"lodash.isfunction": "^3.0.8", "lodash.isfunction": "^3.0.8",
"pre-commit": "^1.2.2", "pre-commit": "^1.2.2",
"pull-stream": "^3.5.0" "pull-stream": "^3.5.0"
}, },
"dependencies": { "dependencies": {
"interface-connection": "~0.3.0", "interface-connection": "~0.3.2",
"mafmt": "^2.1.6", "ip-address": "^5.8.8",
"multiaddr": "^2.2.1",
"ip-address": "^5.8.6",
"lodash.includes": "^4.3.0", "lodash.includes": "^4.3.0",
"lodash.isfunction": "^3.0.8",
"mafmt": "^2.1.8",
"multiaddr": "^2.3.0",
"once": "^1.4.0",
"stream-to-pull-stream": "^1.7.2" "stream-to-pull-stream": "^1.7.2"
}, },
"contributors": [ "contributors": [

View File

@ -6,38 +6,47 @@ const mafmt = require('mafmt')
const includes = require('lodash.includes') const includes = require('lodash.includes')
const isFunction = require('lodash.isfunction') const isFunction = require('lodash.isfunction')
const Connection = require('interface-connection').Connection const Connection = require('interface-connection').Connection
const once = require('once')
const debug = require('debug') const debug = require('debug')
const log = debug('libp2p:tcp:dial') const log = debug('libp2p:tcp:dial')
const createListener = require('./listener') const createListener = require('./listener')
module.exports = class TCP { function noop () {}
dial (ma, options, cb) {
class TCP {
dial (ma, options, callback) {
if (isFunction(options)) { if (isFunction(options)) {
cb = options callback = options
options = {} options = {}
} }
if (!cb) { callback = callback || noop
cb = () => {}
}
callback = once(callback)
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, cb) const rawSocket = net.connect(cOpts)
rawSocket.once('timeout', () => { rawSocket.once('timeout', () => {
log('timeout') log('timeout')
rawSocket.emit('error', new Error('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 socket = toPull.duplex(rawSocket)
const conn = new Connection(socket) const conn = new Connection(socket)
conn.getObservedAddrs = (cb) => { conn.getObservedAddrs = (callback) => {
return cb(null, [ma]) return callback(null, [ma])
} }
return conn return conn
@ -66,3 +75,5 @@ module.exports = class TCP {
}) })
} }
} }
module.exports = TCP

View File

@ -14,17 +14,22 @@ const getMultiaddr = require('./get-multiaddr')
const IPFS_CODE = 421 const IPFS_CODE = 421
const CLOSE_TIMEOUT = 2000 const CLOSE_TIMEOUT = 2000
function noop () {}
module.exports = (handler) => { module.exports = (handler) => {
const listener = new EventEmitter() const listener = new EventEmitter()
const server = net.createServer((socket) => { const server = net.createServer((socket) => {
// Avoid uncaught errors cause by unstable connections
socket.on('error', noop)
const addr = getMultiaddr(socket) const addr = getMultiaddr(socket)
log('new connection', addr.toString()) log('new connection', addr.toString())
const s = toPull.duplex(socket) const s = toPull.duplex(socket)
s.getObservedAddrs = (cb) => { s.getObservedAddrs = (cb) => {
return cb(null, [addr]) cb(null, [addr])
} }
trackSocket(server, socket) trackSocket(server, socket)
@ -34,36 +39,31 @@ module.exports = (handler) => {
listener.emit('connection', conn) listener.emit('connection', conn)
}) })
server.on('listening', () => { server.on('listening', () => listener.emit('listening'))
listener.emit('listening') server.on('error', (err) => listener.emit('error', err))
}) server.on('close', () => listener.emit('close'))
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 // Keep track of open connections to destroy in case of timeout
server.__connections = {} server.__connections = {}
listener.close = (options, cb) => { listener.close = (options, callback) => {
if (typeof options === 'function') { if (typeof options === 'function') {
cb = options callback = options
options = {} options = {}
} }
cb = cb || (() => {}) callback = callback || noop
options = options || {} options = options || {}
let closed = false let closed = false
server.close(cb) server.close(callback)
server.once('close', () => { server.once('close', () => {
closed = true closed = true
}) })
setTimeout(() => { setTimeout(() => {
if (closed) return if (closed) {
return
}
log('unable to close graciously, destroying conns') log('unable to close graciously, destroying conns')
Object.keys(server.__connections).forEach((key) => { Object.keys(server.__connections).forEach((key) => {
@ -76,7 +76,7 @@ module.exports = (handler) => {
let ipfsId let ipfsId
let listeningAddr let listeningAddr
listener.listen = (ma, cb) => { listener.listen = (ma, callback) => {
listeningAddr = ma listeningAddr = ma
if (includes(ma.protoNames(), 'ipfs')) { if (includes(ma.protoNames(), 'ipfs')) {
ipfsId = getIpfsId(ma) ipfsId = getIpfsId(ma)
@ -85,15 +85,15 @@ module.exports = (handler) => {
const lOpts = listeningAddr.toOptions() const lOpts = listeningAddr.toOptions()
log('Listening on %s %s', lOpts.port, lOpts.host) log('Listening on %s %s', lOpts.port, lOpts.host)
return server.listen(lOpts.port, lOpts.host, cb) return server.listen(lOpts.port, lOpts.host, callback)
} }
listener.getAddrs = (cb) => { listener.getAddrs = (callback) => {
const multiaddrs = [] const multiaddrs = []
const address = server.address() const address = server.address()
if (!address) { if (!address) {
return cb(new Error('Listener is not ready yet')) return callback(new Error('Listener is not ready yet'))
} }
// Because TCP will only return the IPv6 version // Because TCP will only return the IPv6 version
@ -128,7 +128,7 @@ module.exports = (handler) => {
multiaddrs.push(ma) multiaddrs.push(ma)
} }
cb(null, multiaddrs) callback(null, multiaddrs)
} }
return listener return listener

View File

@ -2,7 +2,10 @@
'use strict' 'use strict'
const pull = require('pull-stream') const pull = require('pull-stream')
const expect = require('chai').expect const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)
const TCP = require('../src') const TCP = require('../src')
const net = require('net') const net = require('net')
const multiaddr = require('multiaddr') const multiaddr = require('multiaddr')
@ -11,7 +14,7 @@ const Connection = require('interface-connection').Connection
describe('instantiate the transport', () => { describe('instantiate the transport', () => {
it('create', () => { it('create', () => {
const tcp = new TCP() const tcp = new TCP()
expect(tcp).to.exist expect(tcp).to.exist()
}) })
}) })
@ -71,7 +74,7 @@ describe('listen', () => {
const listener = tcp.createListener((conn) => {}) const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => { listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => { listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(multiaddrs.length).to.equal(1) expect(multiaddrs.length).to.equal(1)
expect(multiaddrs[0]).to.deep.equal(mh) expect(multiaddrs[0]).to.deep.equal(mh)
listener.close(done) listener.close(done)
@ -84,7 +87,7 @@ describe('listen', () => {
const listener = tcp.createListener((conn) => {}) const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => { listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => { listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(multiaddrs.length).to.equal(1) expect(multiaddrs.length).to.equal(1)
listener.close(done) listener.close(done)
}) })
@ -96,7 +99,7 @@ describe('listen', () => {
const listener = tcp.createListener((conn) => {}) const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => { listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => { listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(multiaddrs.length > 0).to.equal(true) expect(multiaddrs.length > 0).to.equal(true)
expect(multiaddrs[0].toString().indexOf('0.0.0.0')).to.equal(-1) expect(multiaddrs[0].toString().indexOf('0.0.0.0')).to.equal(-1)
listener.close(done) listener.close(done)
@ -109,7 +112,7 @@ describe('listen', () => {
const listener = tcp.createListener((conn) => {}) const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => { listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => { listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(multiaddrs.length > 0).to.equal(true) expect(multiaddrs.length > 0).to.equal(true)
expect(multiaddrs[0].toString().indexOf('0.0.0.0')).to.equal(-1) expect(multiaddrs[0].toString().indexOf('0.0.0.0')).to.equal(-1)
listener.close(done) listener.close(done)
@ -122,7 +125,7 @@ describe('listen', () => {
const listener = tcp.createListener((conn) => {}) const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => { listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => { listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(multiaddrs.length).to.equal(1) expect(multiaddrs.length).to.equal(1)
expect(multiaddrs[0]).to.deep.equal(mh) expect(multiaddrs[0]).to.deep.equal(mh)
listener.close(done) listener.close(done)
@ -157,12 +160,8 @@ describe('dial', () => {
pull.values(['hey']), pull.values(['hey']),
tcp.dial(ma), tcp.dial(ma),
pull.collect((err, values) => { pull.collect((err, values) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect( expect(values).to.eql([new Buffer('hey!')])
values
).to.be.eql(
[new Buffer('hey!')]
)
done() done()
}) })
) )
@ -173,7 +172,7 @@ describe('dial', () => {
pull( pull(
tcp.dial(ma), tcp.dial(ma),
pull.onEnd((err) => { pull.onEnd((err) => {
expect(err).to.exist expect(err).to.exist()
done() done()
}) })
) )
@ -189,13 +188,9 @@ describe('dial', () => {
pull.values(['hey']), pull.values(['hey']),
tcp.dial(ma), tcp.dial(ma),
pull.collect((err, values) => { pull.collect((err, values) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect( expect(values).to.be.eql([new Buffer('hey')])
values
).to.be.eql([
new Buffer('hey')
])
listener.close(done) listener.close(done)
}) })
@ -258,7 +253,7 @@ describe('dial', () => {
pull.values(['hey']), pull.values(['hey']),
conn, conn,
pull.collect((err, res) => { pull.collect((err, res) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(res).to.be.eql([new Buffer('hey!')]) expect(res).to.be.eql([new Buffer('hey!')])
done() done()
}) })
@ -307,9 +302,9 @@ describe('valid Connection', () => {
let dialerObsAddrs let dialerObsAddrs
const listener = tcp.createListener((conn) => { const listener = tcp.createListener((conn) => {
expect(conn).to.exist expect(conn).to.exist()
conn.getObservedAddrs((err, addrs) => { conn.getObservedAddrs((err, addrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
dialerObsAddrs = addrs dialerObsAddrs = addrs
pull(pull.empty(), conn) pull(pull.empty(), conn)
}) })
@ -324,7 +319,7 @@ describe('valid Connection', () => {
function endHandler () { function endHandler () {
conn.getObservedAddrs((err, addrs) => { conn.getObservedAddrs((err, addrs) => {
expect(err).to.not.exist expect(err).to.not.exist()
pull(pull.empty(), conn) pull(pull.empty(), conn)
closeAndAssert(listener, addrs) closeAndAssert(listener, addrs)
}) })
@ -342,10 +337,10 @@ describe('valid Connection', () => {
it('get Peer Info', (done) => { it('get Peer Info', (done) => {
const listener = tcp.createListener((conn) => { const listener = tcp.createListener((conn) => {
expect(conn).to.exist expect(conn).to.exist()
conn.getPeerInfo((err, peerInfo) => { conn.getPeerInfo((err, peerInfo) => {
expect(err).to.exist expect(err).to.exist()
expect(peerInfo).to.not.exist expect(peerInfo).to.not.exist()
pull(pull.empty(), conn) pull(pull.empty(), conn)
}) })
}) })
@ -356,8 +351,8 @@ describe('valid Connection', () => {
pull(conn, pull.onEnd(endHandler)) pull(conn, pull.onEnd(endHandler))
function endHandler () { function endHandler () {
conn.getPeerInfo((err, peerInfo) => { conn.getPeerInfo((err, peerInfo) => {
expect(err).to.exist expect(err).to.exist()
expect(peerInfo).to.not.exist expect(peerInfo).to.not.exist()
listener.close(done) listener.close(done)
}) })
@ -367,10 +362,10 @@ describe('valid Connection', () => {
it('set Peer Info', (done) => { it('set Peer Info', (done) => {
const listener = tcp.createListener((conn) => { const listener = tcp.createListener((conn) => {
expect(conn).to.exist expect(conn).to.exist()
conn.setPeerInfo('batatas') conn.setPeerInfo('batatas')
conn.getPeerInfo((err, peerInfo) => { conn.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(peerInfo).to.equal('batatas') expect(peerInfo).to.equal('batatas')
pull(pull.empty(), conn) pull(pull.empty(), conn)
}) })
@ -383,7 +378,7 @@ describe('valid Connection', () => {
function endHandler () { function endHandler () {
conn.setPeerInfo('arroz') conn.setPeerInfo('arroz')
conn.getPeerInfo((err, peerInfo) => { conn.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(peerInfo).to.equal('arroz') expect(peerInfo).to.equal('arroz')
listener.close(done) listener.close(done)
@ -424,11 +419,11 @@ describe('Connection wrap', () => {
pull.values(['hey']), pull.values(['hey']),
connWrap, connWrap,
pull.collect((err, chunks) => { pull.collect((err, chunks) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')]) expect(chunks).to.be.eql([new Buffer('hey')])
connWrap.getPeerInfo((err, peerInfo) => { connWrap.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(peerInfo).to.equal('peerInfo') expect(peerInfo).to.equal('peerInfo')
done() done()
}) })
@ -443,7 +438,7 @@ describe('Connection wrap', () => {
pull.values(['hey']), pull.values(['hey']),
connWrap, connWrap,
pull.collect((err, chunks) => { pull.collect((err, chunks) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')]) expect(chunks).to.be.eql([new Buffer('hey')])
done() done()
}) })
@ -459,23 +454,30 @@ describe('Connection wrap', () => {
callback(null, 'none') callback(null, 'none')
} }
conn.getPeerInfo((err, peerInfo) => { conn.getPeerInfo((err, peerInfo) => {
expect(err).to.exist expect(err).to.exist()
}) })
connWrap.getPeerInfo((err, peerInfo) => { connWrap.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(peerInfo).to.equal('none') expect(peerInfo).to.equal('none')
}) })
pull( pull(
pull.values(['hey']), pull.values(['hey']),
connWrap, connWrap,
pull.collect((err, chunks) => { pull.collect((err, chunks) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')]) expect(chunks).to.be.eql([new Buffer('hey')])
done() done()
}) })
) )
}) })
it('dial error', (done) => {
tcp.dial(multiaddr('/ip4/999.0.0.1/tcp/1234'), (err) => {
expect(err).to.exist()
done()
})
})
it('matryoshka wrap', (done) => { it('matryoshka wrap', (done) => {
const conn = tcp.dial(ma) const conn = tcp.dial(ma)
const connWrap1 = new Connection(conn) const connWrap1 = new Connection(conn)
@ -489,10 +491,10 @@ describe('Connection wrap', () => {
pull.values(['hey']), pull.values(['hey']),
connWrap3, connWrap3,
pull.collect((err, chunks) => { pull.collect((err, chunks) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')]) expect(chunks).to.be.eql([new Buffer('hey')])
connWrap3.getPeerInfo((err, peerInfo) => { connWrap3.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist expect(err).to.not.exist()
expect(peerInfo).to.equal('inner doll') expect(peerInfo).to.equal('inner doll')
done() done()
}) })