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

Commit

Permalink
feat(pull): migration to pull-streams
Browse files Browse the repository at this point in the history
  • Loading branch information
@dignifiedquire @daviddias
dignifiedquire authored and daviddias committed Sep 5, 2016
1 parent caa8d6d commit 5e89a26
Show file tree
Hide file tree
Showing 7 changed files with 348 additions and 417 deletions.
37 changes: 22 additions & 15 deletions README.md
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
js-libp2p-tcp
===============
# js-libp2p-tcp

[![](https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square)](http://ipn.io)
[![](https://img.shields.io/badge/freenode-%23ipfs-blue.svg?style=flat-square)](http://webchat.freenode.net/?channels=%23ipfs)
Expand All @@ -21,39 +20,47 @@ js-libp2p-tcp
`multiaddr`. This small shim will enable libp2p to use other different
transports.

**Note:** This module uses [pull-streams](https://pull-stream.github.io) for all stream based interfaces.

## Example

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

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

const tcp = new TCP()

var listener = tcp.createListener(mh1, function handler (socket) {
console.log('connection')
socket.end('bye')
const listener = tcp.createListener(mh1, (socket) => {
console.log('new connection opened')
pull(
pull.values(['hello']),
socket
)
})

listener.listen(mh1, function ready () {
console.log('ready')
listener.listen(() => {
console.log('listening')

const client = tcp.dial(mh1)
client.pipe(process.stdout)
client.on('end', () => {
listener.close()
})
pull(
tcp.dial(mh1),
pull.log,
pull.onEnd(() => {
tcp.close()
})
)
})
```

outputs

```
ready
connection
bye
listening
new connection opened
hello
```

## Installation
Expand Down
14 changes: 7 additions & 7 deletions package.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
"name": "libp2p-tcp",
"version": "0.7.4",
"description": "Node.js implementation of the TCP module that libp2p uses, which implements the interface-connection and interface-transport interfaces",
"main": "lib/index.js",
"main": "src/index.js",
"jsnext:main": "src/index.js",
"scripts": {
"lint": "aegir-lint",
Expand Down Expand Up @@ -32,19 +32,19 @@
},
"homepage": "https://github.com/diasdavid/js-libp2p-tcp",
"devDependencies": {
"aegir": "^4.0.0",
"aegir": "^6.0.0",
"chai": "^3.5.0",
"interface-transport": "^0.2.0",
"pre-commit": "^1.1.2",
"tape": "^4.5.1"
"lodash.isfunction": "^3.0.8",
"pre-commit": "^1.1.2"
},
"dependencies": {
"interface-connection": "0.1.8",
"ip-address": "^5.8.0",
"lodash.contains": "^2.4.3",
"mafmt": "^2.1.2",
"multiaddr": "^2.0.2",
"run-parallel": "^1.1.6"
"pull": "^2.1.1",
"stream-to-pull-stream": "^1.7.0"
},
"contributors": [
"David Dias <daviddias.p@gmail.com>",
Expand All @@ -53,4 +53,4 @@
"Stephen Whitmore <stephen.whitmore@gmail.com>",
"dignifiedquire <dignifiedquire@gmail.com>"
]
}
}
22 changes: 22 additions & 0 deletions src/get-multiaddr.js
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
'use strict'

const multiaddr = require('multiaddr')
const Address6 = require('ip-address').Address6

module.exports = (socket) => {
var mh

if (socket.remoteFamily === 'IPv6') {
var addr = new Address6(socket.remoteAddress)
if (addr.v4) {
var ip4 = addr.to4().correctForm()
mh = multiaddr('/ip4/' + ip4 + '/tcp/' + socket.remotePort)
} else {
mh = multiaddr('/ip6/' + socket.remoteAddress + '/tcp/' + socket.remotePort)
}
} else {
mh = multiaddr('/ip4/' + socket.remoteAddress + '/tcp/' + socket.remotePort)
}

return mh
}
210 changes: 21 additions & 189 deletions src/index.js
View file
Original file line number Diff line number Diff line change
@@ -1,198 +1,48 @@
'use strict'

const debug = require('debug')
const log = debug('libp2p:tcp')
const tcp = require('net')
const multiaddr = require('multiaddr')
const Address6 = require('ip-address').Address6
const net = require('net')
const toPull = require('stream-to-pull-stream')
const mafmt = require('mafmt')
// const parallel = require('run-parallel')
const contains = require('lodash.contains')
const os = require('os')
const isFunction = require('lodash.isfunction')
const Connection = require('interface-connection').Connection
const debug = require('debug')
const log = debug('libp2p:tcp')

exports = module.exports = TCP

const IPFS_CODE = 421
const CLOSE_TIMEOUT = 2000

function TCP () {
if (!(this instanceof TCP)) {
return new TCP()
}
const createListener = require('./listener')

this.dial = function (ma, options, callback) {
if (typeof options === 'function') {
callback = options
module.exports = class TCP {
dial (ma, options, cb) {
if (isFunction(options)) {
cb = options
options = {}
}

if (!callback) {
callback = function noop () {}
if (!cb) {
cb = () => {}
}

const socket = tcp.connect(ma.toOptions())
const conn = new Connection(socket)

socket.on('timeout', () => {
conn.emit('timeout')
})

socket.once('error', (err) => {
callback(err)
})

socket.on('connect', () => {
callback(null, conn)
conn.emit('connect')
})
const cOpts = ma.toOptions()
log('Connecting to %s %s', cOpts.port, cOpts.host)
const socket = toPull.duplex(net.connect(cOpts, cb))

conn.getObservedAddrs = (cb) => {
socket.getObservedAddrs = (cb) => {
return cb(null, [ma])
}

return conn
return new Connection(socket)
}

this.createListener = (options, handler) => {
if (typeof options === 'function') {
createListener (options, handler) {
if (isFunction(options)) {
handler = options
options = {}
}

const listener = tcp.createServer((socket) => {
const conn = new Connection(socket)

conn.getObservedAddrs = (cb) => {
return cb(null, [getMultiaddr(socket)])
}
handler(conn)
})

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]
listeningMultiaddr = ma.decapsulate('ipfs')
}

listener._listen(listeningMultiaddr.toOptions(), callback)
}

listener._close = listener.close
listener.close = (options, callback) => {
if (typeof options === 'function') {
callback = options
options = {}
}
if (!callback) { callback = function noop () {} }
if (!options) { options = {} }

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()
if (m.toString().indexOf('ip4')) {
m = m.decapsulate('tcp')
m = m.encapsulate('/tcp/' + address.port)
if (ipfsHashId) {
m = m.encapsulate('/ipfs/' + ipfsHashId)
}
freshMultiaddrs.push(m)
}
if (address.family === 'IPv6') {
let mh = multiaddr('/ip6/' + address.address + '/tcp/' + address.port)
if (ipfsHashId) {
mh = mh.encapsulate('/ipfs/' + ipfsHashId)
}
freshMultiaddrs.push(mh)
}
cb()
})
listeners.push(listener)
*/
return createListener(handler)
}

this.filter = (multiaddrs) => {
filter (multiaddrs) {
if (!Array.isArray(multiaddrs)) {
multiaddrs = [multiaddrs]
}
Expand All @@ -204,21 +54,3 @@ function TCP () {
})
}
}

function getMultiaddr (socket) {
var mh

if (socket.remoteFamily === 'IPv6') {
var addr = new Address6(socket.remoteAddress)
if (addr.v4) {
var ip4 = addr.to4().correctForm()
mh = multiaddr('/ip4/' + ip4 + '/tcp/' + socket.remotePort)
} else {
mh = multiaddr('/ip6/' + socket.remoteAddress + '/tcp/' + socket.remotePort)
}
} else {
mh = multiaddr('/ip4/' + socket.remoteAddress + '/tcp/' + socket.remotePort)
}

return mh
}
Loading

0 comments on commit 5e89a26

Please sign in to comment.