Compare commits

...

19 Commits

Author SHA1 Message Date
ef45cc9a34 chore: release version v0.10.2 2017-07-22 14:42:22 -07:00
ce5f023508 chore: update contributors 2017-07-22 14:42:21 -07:00
2f9b1fe7fe chore: update deps 2017-07-22 14:40:25 -07:00
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 96 additions and 80 deletions

View File

@ -1,6 +1,6 @@
{
"name": "libp2p-tcp",
"version": "0.9.2",
"version": "0.10.2",
"description": "Node.js implementation of the TCP module that libp2p uses, which implements the interface-connection and interface-transport interfaces",
"main": "src/index.js",
"scripts": {
@ -33,19 +33,22 @@
"node": ">=4.0.0"
},
"devDependencies": {
"interface-transport": "~0.3.3",
"aegir": "^10.0.0",
"chai": "^3.5.0",
"aegir": "^11.0.2",
"chai": "^4.1.0",
"dirty-chai": "^2.0.1",
"interface-transport": "~0.3.5",
"lodash.isfunction": "^3.0.8",
"pre-commit": "^1.2.2",
"pull-stream": "^3.5.0"
"pull-stream": "^3.6.0"
},
"dependencies": {
"interface-connection": "~0.3.0",
"mafmt": "^2.1.6",
"multiaddr": "^2.2.1",
"ip-address": "^5.8.6",
"interface-connection": "~0.3.2",
"ip-address": "^5.8.8",
"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"
},
"contributors": [

View File

@ -6,38 +6,47 @@ const mafmt = require('mafmt')
const includes = require('lodash.includes')
const isFunction = require('lodash.isfunction')
const Connection = require('interface-connection').Connection
const once = require('once')
const debug = require('debug')
const log = debug('libp2p:tcp:dial')
const createListener = require('./listener')
module.exports = class TCP {
dial (ma, options, cb) {
function noop () {}
class TCP {
dial (ma, options, callback) {
if (isFunction(options)) {
cb = options
callback = options
options = {}
}
if (!cb) {
cb = () => {}
}
callback = callback || noop
callback = once(callback)
const cOpts = ma.toOptions()
log('Connecting to %s %s', cOpts.port, cOpts.host)
const rawSocket = net.connect(cOpts, cb)
const rawSocket = net.connect(cOpts)
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 = (cb) => {
return cb(null, [ma])
conn.getObservedAddrs = (callback) => {
return callback(null, [ma])
}
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 CLOSE_TIMEOUT = 2000
function noop () {}
module.exports = (handler) => {
const listener = new EventEmitter()
const server = net.createServer((socket) => {
// Avoid uncaught errors cause by unstable connections
socket.on('error', noop)
const addr = getMultiaddr(socket)
log('new connection', addr.toString())
const s = toPull.duplex(socket)
s.getObservedAddrs = (cb) => {
return cb(null, [addr])
cb(null, [addr])
}
trackSocket(server, socket)
@ -34,36 +39,31 @@ module.exports = (handler) => {
listener.emit('connection', conn)
})
server.on('listening', () => {
listener.emit('listening')
})
server.on('error', (err) => {
listener.emit('error', err)
})
server.on('close', () => {
listener.emit('close')
})
server.on('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, cb) => {
listener.close = (options, callback) => {
if (typeof options === 'function') {
cb = options
callback = options
options = {}
}
cb = cb || (() => {})
callback = callback || noop
options = options || {}
let closed = false
server.close(cb)
server.close(callback)
server.once('close', () => {
closed = true
})
setTimeout(() => {
if (closed) return
if (closed) {
return
}
log('unable to close graciously, destroying conns')
Object.keys(server.__connections).forEach((key) => {
@ -76,7 +76,7 @@ module.exports = (handler) => {
let ipfsId
let listeningAddr
listener.listen = (ma, cb) => {
listener.listen = (ma, callback) => {
listeningAddr = ma
if (includes(ma.protoNames(), 'ipfs')) {
ipfsId = getIpfsId(ma)
@ -85,15 +85,15 @@ module.exports = (handler) => {
const lOpts = listeningAddr.toOptions()
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 address = server.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
@ -128,7 +128,7 @@ module.exports = (handler) => {
multiaddrs.push(ma)
}
cb(null, multiaddrs)
callback(null, multiaddrs)
}
return listener

View File

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