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

Commit

Permalink
comply with the latest interface-transport and interface-connection spec
Browse files Browse the repository at this point in the history
  • Loading branch information
daviddias committed Jun 15, 2016
1 parent 730a477 commit d70d320
Show file tree
Hide file tree
Showing 4 changed files with 428 additions and 256 deletions.
35 changes: 10 additions & 25 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,18 +24,20 @@ transports.
## Example

```js
const Tcp = require('libp2p-tcp')
const TCP = require('libp2p-tcp')
const multiaddr = require('multiaddr')

const mh1 = multiaddr('/ip4/127.0.0.1/tcp/9090')
const mh2 = multiaddr('/ip6/::/tcp/9092')

const tcp = new Tcp()

tcp.createListener([mh1, mh2], function handler (socket) {
var listener = tcp.createListener(mh1, function handler (socket) {
console.log('connection')
socket.end('bye')
}, function ready () {
})

var listener.listen(function ready () {
console.log('ready')

const client = tcp.dial(mh1)
Expand Down Expand Up @@ -65,31 +67,14 @@ bye

## API

```js
const Tcp = require('libp2p-tcp')
```

### var tcp = new Tcp()

Creates a new TCP object. This does nothing on its own, but provides access to
`dial` and `createListener`.

### tcp.createListener(multiaddrs, handler, ready)

Creates TCP servers that listen on the addresses described in the array
`multiaddrs`. Each connection will call `handler` with a connection stream.
`ready` is called once all servers are listening.

### tcp.dial(multiaddr, options={})
[![](https://raw.githubusercontent.com/diasdavid/interface-transport/master/img/badge.png)](https://github.com/diasdavid/interface-transport)

Connects to the multiaddress `multiaddr` using TCP, returning the socket stream.
If `options.ready` is set to a function, it is called when a connection is
established.
`libp2p-tcp` accepts TCP addresses both IPFS and non IPFS encapsulated addresses, i.e:

### tcp.close(callback)
`/ip4/127.0.0.1/tcp/4001`
`/ip4/127.0.0.1/tcp/4001/ipfs/QmHash`

Closes all the listening TCP servers, calling `callback` once all of them have
been shut down.
Both for dialing and listening.

## License

Expand Down
153 changes: 104 additions & 49 deletions src/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -6,68 +6,144 @@ const tcp = require('net')
const multiaddr = require('multiaddr')
const Address6 = require('ip-address').Address6
const mafmt = require('mafmt')
const parallel = require('run-parallel')
// const parallel = require('run-parallel')
const contains = require('lodash.contains')
const os = require('os')

exports = module.exports = TCP

const IPFS_CODE = 421
const CLOSE_TIMEOUT = 300
const CLOSE_TIMEOUT = 2000

function TCP () {
if (!(this instanceof TCP)) {
return new TCP()
}

const listeners = []

this.dial = function (multiaddr, options) {
/*
this.dial = function (ma, options) {
if (!options) {
options = {}
}
options.ready = options.ready || function noop () {}
const conn = tcp.connect(multiaddr.toOptions(), options.ready)
const conn = tcp.connect(ma.toOptions(), options.ready)
conn.getObservedAddrs = () => {
return [multiaddr]
}
return conn
}
*/

this.createListener = (multiaddrs, handler, callback) => {
if (!Array.isArray(multiaddrs)) {
multiaddrs = [multiaddrs]
this.createListener = (options, handler) => {
if (typeof options === 'function') {
handler = options
options = {}
}

const freshMultiaddrs = []
const listener = tcp.createServer((socket) => {
// TODO update to Connection
socket.getObservedAddrs = () => {
return [getMultiaddr(socket)]
}
handler(socket)
})

parallel(multiaddrs.map((m) => (cb) => {
let ipfsHashId
if (contains(m.protoNames(), 'ipfs')) {
ipfsHashId = m.stringTuples().filter((tuple) => {
let ipfsId
let listeningMultiaddr

listener._listen = listener.listen
listener.listen = (ma, callback) => {
listeningMultiaddr = ma
if (contains(ma.protoNames(), 'ipfs')) {
ipfsId = ma.stringTuples().filter((tuple) => {
if (tuple[0] === IPFS_CODE) {
return true
}
})[0][1]
m = m.decapsulate('ipfs')
listeningMultiaddr = ma.decapsulate('ipfs')
}

const listener = tcp.createServer((conn) => {
conn.getObservedAddrs = () => {
return [getMultiaddr(conn)]
}
handler(conn)
})
listener._listen(listeningMultiaddr.toOptions(), callback)
}

listener.__connections = {}
listener.on('connection', (conn) => {
const key = `${conn.remoteAddress}:${conn.remotePort}`
listener.__connections[key] = conn
listener._close = listener.close
listener.close = (options, callback) => {
if (typeof options === 'function') {
callback = options
options = {}
}
if (!callback) { callback = function noop () {} }
if (!options) { options = {} }

conn.on('close', () => {
delete listener.__connections[key]
let closed = false
listener._close(callback)
listener.once('close', () => {
closed = true
})
setTimeout(() => {
if (closed) {
return
}
log('unable to close graciously, destroying conns')
Object.keys(listener.__connections).forEach((key) => {
log('destroying %s', key)
listener.__connections[key].destroy()
})
}, options.timeout || CLOSE_TIMEOUT)
}

// Keep track of open connections to destroy in case of timeout
listener.__connections = {}
listener.on('connection', (socket) => {
const key = `${socket.remoteAddress}:${socket.remotePort}`
listener.__connections[key] = socket

socket.on('close', () => {
delete listener.__connections[key]
})
})

listener.getAddrs = (callback) => {
const multiaddrs = []
const address = listener.address()

// Because TCP will only return the IPv6 version
// we need to capture from the passed multiaddr
if (listeningMultiaddr.toString().indexOf('ip4') !== -1) {
let m = listeningMultiaddr.decapsulate('tcp')
m = m.encapsulate('/tcp/' + address.port)
if (ipfsId) {
m = m.encapsulate('/ipfs/' + ipfsId)
}

if (m.toString().indexOf('0.0.0.0') !== -1) {
const netInterfaces = os.networkInterfaces()
Object.keys(netInterfaces).forEach((niKey) => {
netInterfaces[niKey].forEach((ni) => {
if (ni.family === 'IPv4') {
multiaddrs.push(multiaddr(m.toString().replace('0.0.0.0', ni.address)))
}
})
})
} else {
multiaddrs.push(m)
}
}

if (address.family === 'IPv6') {
let ma = multiaddr('/ip6/' + address.address + '/tcp/' + address.port)
if (ipfsId) {
ma = ma.encapsulate('/ipfs/' + ipfsId)
}

multiaddrs.push(ma)
}

callback(null, multiaddrs)
}

return listener
/*
listener.listen(m.toOptions(), () => {
// Node.js likes to convert addr to IPv6 (when 0.0.0.0 for e.g)
const address = listener.address()
Expand All @@ -92,28 +168,7 @@ function TCP () {
cb()
})
listeners.push(listener)
}), (err) => {
callback(err, freshMultiaddrs)
})
}

this.close = (callback) => {
log('closing')
if (listeners.length === 0) {
log('Called close with no active listeners')
return callback()
}

parallel(listeners.map((listener) => (cb) => {
setTimeout(() => {
Object.keys(listener.__connections).forEach((key) => {
log('destroying %s', key)
listener.__connections[key].destroy()
})
}, CLOSE_TIMEOUT)

listener.close(cb)
}), callback)
*/
}

this.filter = (multiaddrs) => {
Expand Down
Loading

0 comments on commit d70d320

Please sign in to comment.