Skip to content
This repository has been archived by the owner on Aug 29, 2023. It is now read-only.

Commit

Permalink
fix: catch errors on incomming sockets
Browse files Browse the repository at this point in the history
  • Loading branch information
daviddias committed Apr 7, 2017
1 parent 4694f9d commit e204517
Show file tree
Hide file tree
Showing 3 changed files with 39 additions and 36 deletions.
4 changes: 2 additions & 2 deletions package.json
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,8 @@
"ip-address": "^5.8.6",
"lodash.includes": "^4.3.0",
"lodash.isfunction": "^3.0.8",
"mafmt": "^2.1.7",
"multiaddr": "^2.2.3",
"mafmt": "^2.1.8",
"multiaddr": "^2.3.0",
"once": "^1.4.0",
"stream-to-pull-stream": "^1.7.2"
},
Expand Down
27 changes: 15 additions & 12 deletions src/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -12,40 +12,41 @@ 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

cb = once(cb)
callback = once(callback)
const cOpts = ma.toOptions()
log('Connecting to %s %s', cOpts.port, cOpts.host)

const rawSocket = net.connect(cOpts)

rawSocket.once('timeout', () => {
log('timeout')
rawSocket.emit('error', new Error('Timeout'))
})

rawSocket.once('error', cb)
rawSocket.once('error', callback)

rawSocket.once('connect', () => {
rawSocket.removeListener('error', cb)
cb()
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
Expand Down Expand Up @@ -74,3 +75,5 @@ module.exports = class TCP {
})
}
}

module.exports = TCP
44 changes: 22 additions & 22 deletions src/listener.js
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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) => {
Expand All @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -128,7 +128,7 @@ module.exports = (handler) => {
multiaddrs.push(ma)
}

cb(null, multiaddrs)
callback(null, multiaddrs)
}

return listener
Expand Down

0 comments on commit e204517

Please sign in to comment.