Compare commits

..

73 Commits

Author SHA1 Message Date
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
4b211f5cd6 chore: release version v0.9.2 2017-02-09 08:34:54 -08:00
433188f182 chore: update contributors 2017-02-09 08:34:54 -08:00
f3de45d198 chore: ^ to ~ 2017-02-09 08:33:48 -08:00
2afaed5f2a chore(package): update multiaddr to version 2.1.3 (#62)
https://greenkeeper.io/
2017-01-19 11:05:56 +01:00
308dead134 chore(package): update mafmt to version 2.1.4 (#63)
https://greenkeeper.io/
2017-01-16 17:05:57 +01:00
1c8603b721 chore(package): update mafmt to version 2.1.3 (#61)
https://greenkeeper.io/
2017-01-16 16:18:52 +01:00
c883475a4f chore(package): update aegir to version 9.3.0 (#57)
https://greenkeeper.io/
2016-12-18 11:44:36 +00:00
515c6a32ec chore(package): update ip-address to version 5.8.6 (#60)
https://greenkeeper.io/
2016-12-18 11:38:42 +00:00
a107e1da57 chore(package): update pre-commit to version 1.2.2 (#58)
https://greenkeeper.io/
2016-12-18 11:37:15 +00:00
2ec97b3c8c chore(package): update pre-commit to version 1.2.0 (#53)
https://greenkeeper.io/
2016-12-10 12:02:44 -08:00
cb449a6c5c chore(package): update aegir to version 9.2.1 (#54)
https://greenkeeper.io/
2016-12-10 12:01:37 -08:00
1773efecdf chore(package): update ip-address to version 5.8.4 (#52)
https://greenkeeper.io/
2016-12-08 19:19:52 -08:00
49e23f1961 refactor: use lodash.includes instead of lodash.contains
As `loadash.contains` is deprecated.
2016-12-04 13:28:18 +01:00
10b35b22f6 chore(package): update multiaddr to version 2.1.1 (#47)
https://greenkeeper.io/
2016-11-28 12:02:56 +00:00
5677e12592 Merge pull request #44 from libp2p/greenkeeper-aegir-9.1.1
aegir@9.1.1 breaks build ⚠️
2016-11-07 16:12:35 +01:00
9524e8670a chore(package): update aegir to version 9.1.1
https://greenkeeper.io/
2016-11-04 16:07:36 +01:00
32dc4fa5c6 chore: release version v0.9.1 2016-11-03 08:29:14 +00:00
9d697f66a4 chore: update contributors 2016-11-03 08:29:14 +00:00
fb6c61abe9 chore: update deps 2016-11-03 08:29:09 +00:00
114998db1c chore: release version v0.9.0 2016-11-03 08:23:38 +00:00
4a94e3ca01 chore: update contributors 2016-11-03 08:23:38 +00:00
e451201848 Async Crypto Endeavour (#40) 2016-11-03 08:22:32 +00:00
e7524e1747 docs(readme): fix coverage badge 2016-09-09 12:13:14 +02:00
06689e3ff0 fix(deps): remove unused pull dep 2016-09-09 11:49:11 +02:00
376f938cc0 Merge pull request #34 from libp2p/greenkeeper-aegir-8.0.1
aegir@8.0.1 breaks build ⚠️
2016-09-08 20:07:25 -04:00
a2958b9bca chore(package): update aegir to version 8.0.1
https://greenkeeper.io/
2016-09-09 01:34:28 +02:00
7b7e521e57 Merge pull request #33 from libp2p/greenkeeper-multiaddr-2.0.3
multiaddr@2.0.3 breaks build ⚠️
2016-09-07 20:21:13 -04:00
2330732ed5 chore(package): update multiaddr to version 2.0.3
https://greenkeeper.io/
2016-09-08 01:20:22 +02:00
7b7e7f6344 Merge pull request #31 from libp2p/greenkeeper-aegir-8.0.0
Update aegir to version 8.0.0 🚀
2016-09-07 14:13:02 -04:00
850a414e6f chore: release version v0.8.1 2016-09-06 09:31:40 -04:00
f6df6cfa40 chore: update contributors 2016-09-06 09:31:40 -04:00
c16376f567 Merge pull request #32 from libp2p/feature/standardize-readme
Standardize Readme
2016-09-06 09:24:19 -04:00
0025407a5b Standardize Readme
Also did some edits to the text here and there. Added a contribute section, too.
2016-09-06 09:16:32 -04:00
48708ac4d2 chore(package): update aegir to version 8.0.0
https://greenkeeper.io/
2016-09-06 15:11:49 +02:00
1d1d091082 chore: release version v0.8.0 2016-09-06 08:52:42 -04:00
0d007ff5fe chore: update contributors 2016-09-06 08:52:42 -04:00
2f2ca0d97b Merge pull request #20 from libp2p/pull
Migrate to pull-streams
2016-09-06 08:42:29 -04:00
da8ee216e1 feat(deps): update to published deps 2016-09-06 08:31:17 -04:00
d9f65e0b0c feat(readme): add pull-streams documentation 2016-09-05 18:31:34 -04:00
5e89a2608b feat(pull): migration to pull-streams 2016-09-05 18:31:34 -04:00
caa8d6dbb1 Merge pull request #18 from emschwartz/master
docs: fix README example
2016-09-05 18:01:32 -04:00
9b61bbc040 Merge pull request #23 from libp2p/greenkeeper-mafmt-2.1.2
mafmt@2.1.2 breaks build ⚠️
2016-08-31 06:49:48 -04:00
915fdc818f chore(package): update mafmt to version 2.1.2
https://greenkeeper.io/
2016-08-22 16:19:30 +02:00
2dc52d0a5c docs: fix README example 2016-08-03 17:52:01 +02:00
e01046eb03 chore: release version v0.7.4 2016-08-03 14:57:30 +01:00
918b9df322 chore: update contributors 2016-08-03 14:57:30 +01:00
31b47fbfd2 Merge pull request #17 from libp2p/update-deps
update dependencies
2016-08-03 14:47:05 +01:00
26e9268bb5 update dependencies 2016-08-03 14:42:35 +01:00
676b5f3dbf chore: release version v0.7.3 2016-06-26 21:18:26 +01:00
ad7a331d9a chore: update contributors 2016-06-26 21:18:25 +01:00
a00c38437e Merge pull request #16 from libp2p/destroy-properly
use duplexify fork while PR is not up yet
2016-06-26 21:18:02 +01:00
cf0e9a1a8f use duplexify fork while PR is not up yet 2016-06-26 21:13:39 +01:00
d2f91814b6 chore: release version v0.7.2 2016-06-23 08:42:03 +01:00
405f552c77 chore: update contributors 2016-06-23 08:42:02 +01:00
cba4b8b94a pass the err on the callback too 2016-06-23 08:41:54 +01:00
ae6a808fc7 chore: release version v0.7.1 2016-06-23 08:35:56 +01:00
eae6819b02 chore: update contributors 2016-06-23 08:35:56 +01:00
9ac5cca946 fix: error was passed in duplicate 2016-06-23 08:35:49 +01:00
cdfd473cac chore: release version v0.7.0 2016-06-22 11:31:32 +01:00
5344ea33aa chore: update contributors 2016-06-22 11:31:32 +01:00
c2654206d3 Merge pull request #15 from libp2p/follow-spec
comply with the latest interface-transport and interface-connection spec
2016-06-22 11:28:15 +01:00
539a007031 comply with the latest interface-transport and interface-connection spec 2016-06-22 09:36:08 +01:00
730a477bd8 chore: release version v0.6.2 2016-06-01 02:15:22 +02:00
19b29fe40b chore: update contributors 2016-06-01 02:15:22 +02:00
073e14553b Merge pull request #14 from diasdavid/close-timeout
fix: destroy hanging connections after timeout
2016-06-01 01:09:40 +01:00
2ed01e8f5b fix: address cr 2016-05-31 21:49:12 +02:00
4a121696d1 fix: destroy hanging connections after timeout 2016-05-31 14:48:18 +02:00
11 changed files with 837 additions and 330 deletions

3
.gitignore vendored
View File

@ -25,6 +25,3 @@ build/Release
# Dependency directory
# https://www.npmjs.org/doc/misc/npm-faq.html#should-i-check-my-node_modules-folder-into-git
node_modules
lib
dist

View File

@ -1,9 +1,13 @@
sudo: false
language: node_js
node_js:
- 4
- 5
- stable
matrix:
include:
- node_js: 4
env: CXX=g++-4.8
- node_js: 6
env: CXX=g++-4.8
- node_js: stable
env: CXX=g++-4.8
# Make sure we have new NPM.
before_install:
@ -14,12 +18,16 @@ script:
- npm test
- npm run coverage
addons:
firefox: 'latest'
before_script:
- export DISPLAY=:99.0
- sh -e /etc/init.d/xvfb start
after_success:
- npm run coverage-publish
addons:
apt:
sources:
- ubuntu-toolchain-r-test
packages:
- g++-4.8

View File

@ -1,6 +1,6 @@
The MIT License (MIT)
Copyright (c) 2015 David Dias
Copyright (c) 2015-2016 David Dias
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

165
README.md
View File

@ -1,61 +1,36 @@
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)
[![Build Status](https://travis-ci.org/diasdavid/js-libp2p-tcp.svg?style=flat-square)](https://travis-ci.org/diasdavid/js-libp2p-tcp)
![](https://img.shields.io/badge/coverage-%3F-yellow.svg?style=flat-square)
[![Dependency Status](https://david-dm.org/diasdavid/js-libp2p-tcp.svg?style=flat-square)](https://david-dm.org/diasdavid/js-libp2p-tcp)
[![Build Status](https://travis-ci.org/libp2p/js-libp2p-tcp.svg?style=flat-square)](https://travis-ci.org/libp2p/js-libp2p-tcp)
[![Coverage Status](https://coveralls.io/repos/github/libp2p/js-libp2p-tcp/badge.svg?branch=master)](https://coveralls.io/github/libp2p/js-libp2p-tcp?branch=master)
[![Dependency Status](https://david-dm.org/libp2p/js-libp2p-tcp.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-tcp)
[![js-standard-style](https://img.shields.io/badge/code%20style-standard-brightgreen.svg?style=flat-square)](https://github.com/feross/standard)
![](https://img.shields.io/badge/npm-%3E%3D3.0.0-orange.svg?style=flat-square)
![](https://img.shields.io/badge/Node.js-%3E%3D4.0.0-orange.svg?style=flat-square)
![](https://raw.githubusercontent.com/diasdavid/abstract-connection/master/img/badge.png)
![](https://raw.githubusercontent.com/diasdavid/abstract-transport/master/img/badge.png)
![](https://raw.githubusercontent.com/libp2p/interface-connection/master/img/badge.png)
![](https://raw.githubusercontent.com/libp2p/interface-transport/master/img/badge.png)
> Node.js implementation of the TCP module that libp2p uses, which implements
> the [interface-connection](https://github.com/diasdavid/interface-connection)
> interface for dial/listen.
> Node.js implementation of the TCP module that libp2p uses, which implements the [interface-connection](https://github.com/libp2p/interface-connection) interface for dial/listen.
## Description
`libp2p-tcp` in Node.js is a very thin shim that adds support for dialing to a `multiaddr`. This small shim will enable libp2p to use other different transports.
`libp2p-tcp` in Node.js is a very thin shim that adds support for dialing to a
`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
## Table of Contents
```js
const Tcp = require('libp2p-tcp')
const multiaddr = require('multiaddr')
- [Install](#install)
- [npm](#npm)
- [Usage](#usage)
- [Example](#example)
- [This module uses `pull-streams`](#this-module-uses-pull-streams)
- [Converting `pull-streams` to Node.js Streams](#converting-pull-streams-to-nodejs-streams)
- [API](#api)
- [Contribute](#contribute)
- [License](#license)
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) {
console.log('connection')
socket.end('bye')
}, function ready () {
console.log('ready')
const client = tcp.dial(mh1)
client.pipe(process.stdout)
client.on('end', () => {
tcp.close()
})
})
```
outputs
```
ready
connection
bye
```
## Installation
## Install
### npm
@ -63,34 +38,96 @@ bye
> npm i libp2p-tcp
```
## API
## Usage
### Example
```js
const Tcp = require('libp2p-tcp')
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()
const listener = tcp.createListener(mh1, (socket) => {
console.log('new connection opened')
pull(
pull.values(['hello']),
socket
)
})
listener.listen(() => {
console.log('listening')
pull(
tcp.dial(mh1),
pull.log,
pull.onEnd(() => {
tcp.close()
})
)
})
```
### var tcp = new Tcp()
Outputs:
Creates a new TCP object. This does nothing on its own, but provides access to
`dial` and `createListener`.
```sh
listening
new connection opened
hello
```
### tcp.createListener(multiaddrs, handler, ready)
### This module uses `pull-streams`
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.
We expose a streaming interface based on `pull-streams`, rather then on the Node.js core streams implementation (aka Node.js streams). `pull-streams` offers us a better mechanism for error handling and flow control guarantees. If you would like to know more about why we did this, see the discussion at this [issue](https://github.com/ipfs/js-ipfs/issues/362).
### tcp.dial(multiaddr, options={})
You can learn more about pull-streams at:
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.
- [The history of Node.js streams, nodebp April 2014](https://www.youtube.com/watch?v=g5ewQEuXjsQ)
- [The history of streams, 2016](http://dominictarr.com/post/145135293917/history-of-streams)
- [pull-streams, the simple streaming primitive](http://dominictarr.com/post/149248845122/pull-streams-pull-streams-are-a-very-simple)
- [pull-streams documentation](https://pull-stream.github.io/)
### tcp.close(callback)
#### Converting `pull-streams` to Node.js Streams
Closes all the listening TCP servers, calling `callback` once all of them have
been shut down.
If you are a Node.js streams user, you can convert a pull-stream to a Node.js stream using the module [`pull-stream-to-stream`](https://github.com/dominictarr/pull-stream-to-stream), giving you an instance of a Node.js stream that is linked to the pull-stream. For example:
```js
const pullToStream = require('pull-stream-to-stream')
const nodeStreamInstance = pullToStream(pullStreamInstance)
// nodeStreamInstance is an instance of a Node.js Stream
```
To learn more about this utility, visit https://pull-stream.github.io/#pull-stream-to-stream.
## API
[![](https://raw.githubusercontent.com/libp2p/interface-transport/master/img/badge.png)](https://github.com/libp2p/interface-transport)
`libp2p-tcp` accepts TCP addresses both IPFS and non IPFS encapsulated addresses, i.e:
`/ip4/127.0.0.1/tcp/4001`
`/ip4/127.0.0.1/tcp/4001/ipfs/QmHash`
Both for dialing and listening.
## Contribute
Contributions are welcome! The libp2p implementation in JavaScript is a work in progress. As such, there's a few things you can do right now to help out:
- [Check out the existing issues](//github.com/libp2p/js-libp2p-tcp/issues).
- **Perform code reviews**.
- **Add tests**. There can never be enough tests.
Please be aware that all interactions related to libp2p are subject to the IPFS [Code of Conduct](https://github.com/ipfs/community/blob/master/code-of-conduct.md).
Small note: If editing the README, please conform to the [standard-readme](https://github.com/RichardLitt/standard-readme) specification.
## License
MIT
[MIT](LICENSE) © 2015-2016 David Dias

View File

@ -1,16 +1,14 @@
{
"name": "libp2p-tcp",
"version": "0.6.1",
"version": "0.9.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",
"jsnext:main": "src/index.js",
"main": "src/index.js",
"scripts": {
"lint": "aegir-lint",
"build": "aegir-build --env node",
"test": "aegir-test --env node",
"release": "aegir-release --env node",
"release-minor": "aegir-release --type minor --env node",
"release-major": "aegir-release --type major --env node",
"release": "aegir-release --env no-build",
"release-minor": "aegir-release --type minor --env no-build",
"release-major": "aegir-release --type major --env no-build",
"coverage": "aegir-coverage",
"coverage-publish": "aegir-coverage publish"
},
@ -20,7 +18,7 @@
],
"repository": {
"type": "git",
"url": "https://github.com/diasdavid/js-libp2p-tcp.git"
"url": "https://github.com/libp2p/js-libp2p-tcp.git"
},
"keywords": [
"IPFS"
@ -28,28 +26,37 @@
"author": "David Dias <daviddias@ipfs.io>",
"license": "MIT",
"bugs": {
"url": "https://github.com/diasdavid/js-libp2p-tcp/issues"
"url": "https://github.com/libp2p/js-libp2p-tcp/issues"
},
"homepage": "https://github.com/libp2p/js-libp2p-tcp",
"engines": {
"node": ">=4.0.0"
},
"homepage": "https://github.com/diasdavid/js-libp2p-tcp",
"devDependencies": {
"aegir": "^3.0.4",
"aegir": "^11.0.0",
"chai": "^3.5.0",
"interface-connection": "0.0.3",
"interface-transport": "^0.1.1",
"pre-commit": "^1.1.2",
"tape": "^4.5.1"
"dirty-chai": "^1.2.2",
"interface-transport": "~0.3.5",
"lodash.isfunction": "^3.0.8",
"pre-commit": "^1.2.2",
"pull-stream": "^3.5.0"
},
"dependencies": {
"ip-address": "^5.8.0",
"lodash.contains": "^2.4.3",
"mafmt": "^2.1.0",
"multiaddr": "^2.0.2",
"run-parallel": "^1.1.6"
"interface-connection": "~0.3.2",
"ip-address": "^5.8.6",
"lodash.includes": "^4.3.0",
"lodash.isfunction": "^3.0.8",
"mafmt": "^2.1.6",
"multiaddr": "^2.2.2",
"stream-to-pull-stream": "^1.7.2"
},
"contributors": [
"David Dias <daviddias.p@gmail.com>",
"Evan Schwartz <evan.mark.schwartz@gmail.com>",
"Friedel Ziegelmayer <dignifiedquire@gmail.com>",
"Greenkeeper <support@greenkeeper.io>",
"João Antunes <j.goncalo.antunes@gmail.com>",
"Prashanth Chandra <coolshanth94@gmail.com>",
"Richard Littauer <richard.littauer@gmail.com>",
"Stephen Whitmore <stephen.whitmore@gmail.com>"
]

22
src/get-multiaddr.js Normal file
View File

@ -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
}

View File

@ -1,128 +1,68 @@
'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 includes = require('lodash.includes')
const isFunction = require('lodash.isfunction')
const Connection = require('interface-connection').Connection
const debug = require('debug')
const log = debug('libp2p:tcp:dial')
exports = module.exports = TCP
const createListener = require('./listener')
const IPFS_CODE = 421
function TCP () {
if (!(this instanceof TCP)) {
return new TCP()
}
const listeners = []
this.dial = function (multiaddr, options) {
if (!options) {
module.exports = class TCP {
dial (ma, options, cb) {
if (isFunction(options)) {
cb = options
options = {}
}
options.ready = options.ready || function noop () {}
const conn = tcp.connect(multiaddr.toOptions(), options.ready)
conn.getObservedAddrs = () => {
return [multiaddr]
if (!cb) {
cb = () => {}
}
const cOpts = ma.toOptions()
log('Connecting to %s %s', cOpts.port, cOpts.host)
const rawSocket = net.connect(cOpts, cb)
rawSocket.once('timeout', () => {
log('timeout')
rawSocket.emit('error', new Error('Timeout'))
})
const socket = toPull.duplex(rawSocket)
const conn = new Connection(socket)
conn.getObservedAddrs = (cb) => {
return cb(null, [ma])
}
return conn
}
this.createListener = (multiaddrs, handler, callback) => {
if (!Array.isArray(multiaddrs)) {
multiaddrs = [multiaddrs]
createListener (options, handler) {
if (isFunction(options)) {
handler = options
options = {}
}
const freshMultiaddrs = []
handler = handler || (() => {})
parallel(multiaddrs.map((m) => (cb) => {
let ipfsHashId
if (contains(m.protoNames(), 'ipfs')) {
ipfsHashId = m.stringTuples().filter((tuple) => {
if (tuple[0] === IPFS_CODE) {
return true
}
})[0][1]
m = m.decapsulate('ipfs')
}
const listener = tcp.createServer((conn) => {
conn.getObservedAddrs = () => {
return [getMultiaddr(conn)]
}
handler(conn)
})
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)
}), (err) => {
callback(err, freshMultiaddrs)
})
return createListener(handler)
}
this.close = (callback) => {
if (listeners.length === 0) {
log('Called close with no active listeners')
return callback()
}
parallel(listeners.map((listener) => {
return (cb) => listener.close(cb)
}), callback)
}
this.filter = (multiaddrs) => {
filter (multiaddrs) {
if (!Array.isArray(multiaddrs)) {
multiaddrs = [multiaddrs]
}
return multiaddrs.filter((ma) => {
if (contains(ma.protoNames(), 'ipfs')) {
if (includes(ma.protoNames(), 'ipfs')) {
ma = ma.decapsulate('ipfs')
}
return mafmt.TCP.matches(ma)
})
}
}
function getMultiaddr (conn) {
var mh
if (conn.remoteFamily === 'IPv6') {
var addr = new Address6(conn.remoteAddress)
if (addr.v4) {
var ip4 = addr.to4().correctForm()
mh = multiaddr('/ip4/' + ip4 + '/tcp/' + conn.remotePort)
} else {
mh = multiaddr('/ip6/' + conn.remoteAddress + '/tcp/' + conn.remotePort)
}
} else {
mh = multiaddr('/ip4/' + conn.remoteAddress + '/tcp/' + conn.remotePort)
}
return mh
}

150
src/listener.js Normal file
View File

@ -0,0 +1,150 @@
'use strict'
const multiaddr = require('multiaddr')
const Connection = require('interface-connection').Connection
const os = require('os')
const includes = require('lodash.includes')
const net = require('net')
const toPull = require('stream-to-pull-stream')
const EventEmitter = require('events').EventEmitter
const debug = require('debug')
const log = debug('libp2p:tcp:listen')
const getMultiaddr = require('./get-multiaddr')
const IPFS_CODE = 421
const CLOSE_TIMEOUT = 2000
module.exports = (handler) => {
const listener = new EventEmitter()
const server = net.createServer((socket) => {
const addr = getMultiaddr(socket)
log('new connection', addr.toString())
const s = toPull.duplex(socket)
s.getObservedAddrs = (cb) => {
return cb(null, [addr])
}
trackSocket(server, socket)
const conn = new Connection(s)
handler(conn)
listener.emit('connection', conn)
})
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) => {
if (typeof options === 'function') {
cb = options
options = {}
}
cb = cb || (() => {})
options = options || {}
let closed = false
server.close(cb)
server.once('close', () => {
closed = true
})
setTimeout(() => {
if (closed) return
log('unable to close graciously, destroying conns')
Object.keys(server.__connections).forEach((key) => {
log('destroying %s', key)
server.__connections[key].destroy()
})
}, options.timeout || CLOSE_TIMEOUT)
}
let ipfsId
let listeningAddr
listener.listen = (ma, cb) => {
listeningAddr = ma
if (includes(ma.protoNames(), 'ipfs')) {
ipfsId = getIpfsId(ma)
listeningAddr = ma.decapsulate('ipfs')
}
const lOpts = listeningAddr.toOptions()
log('Listening on %s %s', lOpts.port, lOpts.host)
return server.listen(lOpts.port, lOpts.host, cb)
}
listener.getAddrs = (cb) => {
const multiaddrs = []
const address = server.address()
if (!address) {
return cb(new Error('Listener is not ready yet'))
}
// Because TCP will only return the IPv6 version
// we need to capture from the passed multiaddr
if (listeningAddr.toString().indexOf('ip4') !== -1) {
let m = listeningAddr.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)
}
cb(null, multiaddrs)
}
return listener
}
function getIpfsId (ma) {
return ma.stringTuples().filter((tuple) => {
return tuple[0] === IPFS_CODE
})[0][1]
}
function trackSocket (server, socket) {
const key = `${socket.remoteAddress}:${socket.remotePort}`
server.__connections[key] = socket
socket.on('close', () => {
delete server.__connections[key]
})
}

23
test/compliance.spec.js Normal file
View File

@ -0,0 +1,23 @@
/* eslint-env mocha */
'use strict'
const tests = require('interface-transport')
const multiaddr = require('multiaddr')
const Tcp = require('../src')
describe('interface-transport compliance', () => {
tests({
setup (cb) {
let tcp = new Tcp()
const addrs = [
multiaddr('/ip4/127.0.0.1/tcp/9091'),
multiaddr('/ip4/127.0.0.1/tcp/9092'),
multiaddr('/ip4/127.0.0.1/tcp/9093')
]
cb(null, tcp, addrs)
},
teardown (cb) {
cb()
}
})
})

View File

@ -1,134 +1,274 @@
/* eslint-env mocha */
'use strict'
const expect = require('chai').expect
const TCPlibp2p = require('../src')
const pull = require('pull-stream')
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')
const Connection = require('interface-connection').Connection
describe('libp2p-tcp', function () {
this.timeout(10000)
var tcp
describe('instantiate the transport', () => {
it('create', () => {
const tcp = new TCP()
expect(tcp).to.exist()
})
})
it('create', (done) => {
tcp = new TCPlibp2p()
expect(tcp).to.exist
done()
describe('listen', () => {
let tcp
beforeEach(() => {
tcp = new TCP()
})
it('create without new', (done) => {
tcp = TCPlibp2p()
expect(tcp).to.exist
done()
it('close listener with connections, through timeout', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/9090/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
const listener = tcp.createListener((conn) => {
pull(conn, conn)
})
listener.listen(mh, () => {
const socket1 = net.connect(9090)
const socket2 = net.connect(9090)
socket1.write('Some data that is never handled')
socket1.end()
socket1.on('error', () => {})
socket2.on('error', () => {})
socket1.on('connect', () => {
listener.close(done)
})
})
})
it('close /wo listeners', (done) => {
tcp = new TCPlibp2p()
expect(tcp).to.exist
expect(function () { tcp.close() }).to.throw(Error)
done()
it('listen on port 0', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/0')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.close(done)
})
})
it('listen', (done) => {
it('listen on IPv6 addr', (done) => {
const mh = multiaddr('/ip6/::/tcp/9090')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.close(done)
})
})
it('listen on any Interface', (done) => {
const mh = multiaddr('/ip4/0.0.0.0/tcp/9090')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.close(done)
})
})
it('getAddrs', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/9090')
tcp.createListener(mh, (socket) => {
expect(socket).to.exist
socket.end()
tcp.close(() => {
done()
})
}, (err, freshMultiaddrs) => {
expect(err).to.not.exist
expect(mh).to.deep.equal(freshMultiaddrs[0])
const socket = net.connect({ host: '127.0.0.1', port: 9090 })
socket.end()
})
})
it('listen on addr with /ipfs/QmHASH', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/14090/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
tcp.createListener(mh, (socket) => {
expect(socket).to.exist
socket.end()
tcp.close(() => {
done()
})
}, (err, freshMultiaddrs) => {
expect(err).to.not.exist
expect(mh).to.deep.equal(freshMultiaddrs[0])
const socket = net.connect({ host: '127.0.0.1', port: 14090 })
socket.end()
})
})
it('dial', (done) => {
const server = net.createServer((socket) => {
expect(socket).to.exist
socket.end()
server.close(done)
})
server.listen(9090, () => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/9090')
const socket = tcp.dial(mh)
socket.end()
})
})
it('listen on several', (done) => {
const mh1 = multiaddr('/ip4/127.0.0.1/tcp/9090')
const mh2 = multiaddr('/ip4/127.0.0.1/tcp/9091')
const mh3 = multiaddr('/ip6/::/tcp/9092')
const tcp = new TCPlibp2p()
tcp.createListener([mh1, mh2, mh3], (socket) => {}, () => {
tcp.close(done)
})
})
it('dial ipv6', (done) => {
const mh = multiaddr('/ip6/::/tcp/9091')
var dialerObsAddrs
tcp.createListener(mh, (conn) => {
expect(conn).to.exist
dialerObsAddrs = conn.getObservedAddrs()
conn.end()
}, () => {
const conn = tcp.dial(mh)
conn.on('end', () => {
expect(dialerObsAddrs.length).to.equal(1)
tcp.close()
done()
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist()
expect(multiaddrs.length).to.equal(1)
expect(multiaddrs[0]).to.deep.equal(mh)
listener.close(done)
})
})
})
it('get observed addrs', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/9090')
var dialerObsAddrs
var listenerObsAddrs
it('getAddrs on port 0 listen', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/0')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist()
expect(multiaddrs.length).to.equal(1)
listener.close(done)
})
})
})
tcp.createListener(mh, (conn) => {
expect(conn).to.exist
dialerObsAddrs = conn.getObservedAddrs()
conn.end()
}, () => {
const conn = tcp.dial(mh)
conn.on('end', () => {
listenerObsAddrs = conn.getObservedAddrs()
conn.end()
it('getAddrs from listening on 0.0.0.0', (done) => {
const mh = multiaddr('/ip4/0.0.0.0/tcp/9090')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => {
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)
})
})
})
tcp.close(() => {
expect(listenerObsAddrs[0]).to.deep.equal(mh)
expect(dialerObsAddrs.length).to.equal(1)
done()
it('getAddrs from listening on 0.0.0.0 and port 0', (done) => {
const mh = multiaddr('/ip4/0.0.0.0/tcp/0')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => {
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)
})
})
})
it('getAddrs preserves IPFS Id', (done) => {
const mh = multiaddr('/ip4/127.0.0.1/tcp/9090/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
const listener = tcp.createListener((conn) => {})
listener.listen(mh, () => {
listener.getAddrs((err, multiaddrs) => {
expect(err).to.not.exist()
expect(multiaddrs.length).to.equal(1)
expect(multiaddrs[0]).to.deep.equal(mh)
listener.close(done)
})
})
})
})
describe('dial', () => {
let tcp
let listener
const ma = multiaddr('/ip4/127.0.0.1/tcp/9090')
beforeEach((done) => {
tcp = new TCP()
listener = tcp.createListener((conn) => {
pull(
conn,
pull.map((x) => new Buffer(x.toString() + '!')),
conn
)
})
listener.listen(ma, done)
})
afterEach((done) => {
listener.close(done)
})
it('dial on IPv4', (done) => {
pull(
pull.values(['hey']),
tcp.dial(ma),
pull.collect((err, values) => {
expect(err).to.not.exist()
expect(values).to.eql([new Buffer('hey!')])
done()
})
)
})
it('dial to non existent listener', (done) => {
const ma = multiaddr('/ip4/127.0.0.1/tcp/8989')
pull(
tcp.dial(ma),
pull.onEnd((err) => {
expect(err).to.exist()
done()
})
)
})
it('dial on IPv6', (done) => {
const ma = multiaddr('/ip6/::/tcp/9066')
const listener = tcp.createListener((conn) => {
pull(conn, conn)
})
listener.listen(ma, () => {
pull(
pull.values(['hey']),
tcp.dial(ma),
pull.collect((err, values) => {
expect(err).to.not.exist()
expect(values).to.be.eql([new Buffer('hey')])
listener.close(done)
})
})
)
})
})
it('filter valid addrs for this transport', (done) => {
it.skip('dial and destroy on listener', (done) => {
// TODO: why is this failing
let count = 0
const closed = ++count === 2 ? finish() : null
const ma = multiaddr('/ip6/::/tcp/9067')
const listener = tcp.createListener((conn) => {
pull(
pull.empty(),
conn,
pull.onEnd(closed)
)
})
listener.listen(ma, () => {
pull(tcp.dial(ma), pull.onEnd(closed))
})
function finish () {
listener.close(done)
}
})
it('dial and destroy on dialer', (done) => {
let count = 0
const destroyed = () => ++count === 2 ? finish() : null
const ma = multiaddr('/ip6/::/tcp/9068')
const listener = tcp.createListener((conn) => {
pull(conn, pull.onEnd(destroyed))
})
listener.listen(ma, () => {
pull(
pull.empty(),
tcp.dial(ma),
pull.onEnd(destroyed)
)
})
function finish () {
listener.close(done)
}
})
it('dial on IPv4 with IPFS Id', (done) => {
const ma = multiaddr('/ip4/127.0.0.1/tcp/9090/ipfs/Qmb6owHp6eaWArVbcJJbQSyifyJBttMMjYV76N2hMbf5Vw')
const conn = tcp.dial(ma)
pull(
pull.values(['hey']),
conn,
pull.collect((err, res) => {
expect(err).to.not.exist()
expect(res).to.be.eql([new Buffer('hey!')])
done()
})
)
})
})
describe('filter addrs', () => {
let tcp
before(() => {
tcp = new TCP()
})
it('filter valid addrs for this transport', () => {
const mh1 = multiaddr('/ip4/127.0.0.1/tcp/9090')
const mh2 = multiaddr('/ip4/127.0.0.1/udp/9090')
const mh3 = multiaddr('/ip4/127.0.0.1/tcp/9090/http')
@ -137,15 +277,221 @@ describe('libp2p-tcp', function () {
const valid = tcp.filter([mh1, mh2, mh3, mh4])
expect(valid.length).to.equal(2)
expect(valid[0]).to.deep.equal(mh1)
done()
expect(valid[1]).to.deep.equal(mh4)
})
it('filter a valid addr for this transport', (done) => {
it('filter a single addr for this transport', () => {
const mh1 = multiaddr('/ip4/127.0.0.1/tcp/9090')
const valid = tcp.filter(mh1)
expect(valid.length).to.equal(1)
expect(valid[0]).to.deep.equal(mh1)
done()
})
})
describe('valid Connection', () => {
let tcp
beforeEach(() => {
tcp = new TCP()
})
const ma = multiaddr('/ip4/127.0.0.1/tcp/9090')
it('get observed addrs', (done) => {
let dialerObsAddrs
const listener = tcp.createListener((conn) => {
expect(conn).to.exist()
conn.getObservedAddrs((err, addrs) => {
expect(err).to.not.exist()
dialerObsAddrs = addrs
pull(pull.empty(), conn)
})
})
listener.listen(ma, () => {
const conn = tcp.dial(ma)
pull(
conn,
pull.onEnd(endHandler)
)
function endHandler () {
conn.getObservedAddrs((err, addrs) => {
expect(err).to.not.exist()
pull(pull.empty(), conn)
closeAndAssert(listener, addrs)
})
}
function closeAndAssert (listener, addrs) {
listener.close(() => {
expect(addrs[0]).to.deep.equal(ma)
expect(dialerObsAddrs.length).to.equal(1)
done()
})
}
})
})
it('get Peer Info', (done) => {
const listener = tcp.createListener((conn) => {
expect(conn).to.exist()
conn.getPeerInfo((err, peerInfo) => {
expect(err).to.exist()
expect(peerInfo).to.not.exist()
pull(pull.empty(), conn)
})
})
listener.listen(ma, () => {
const conn = tcp.dial(ma)
pull(conn, pull.onEnd(endHandler))
function endHandler () {
conn.getPeerInfo((err, peerInfo) => {
expect(err).to.exist()
expect(peerInfo).to.not.exist()
listener.close(done)
})
}
})
})
it('set Peer Info', (done) => {
const listener = tcp.createListener((conn) => {
expect(conn).to.exist()
conn.setPeerInfo('batatas')
conn.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist()
expect(peerInfo).to.equal('batatas')
pull(pull.empty(), conn)
})
})
listener.listen(ma, () => {
const conn = tcp.dial(ma)
pull(conn, pull.onEnd(endHandler))
function endHandler () {
conn.setPeerInfo('arroz')
conn.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist()
expect(peerInfo).to.equal('arroz')
listener.close(done)
})
}
})
})
})
describe.skip('turbolence', () => {
it('dialer - emits error on the other end is terminated abruptly', (done) => {})
it('listener - emits error on the other end is terminated abruptly', (done) => {})
})
describe('Connection wrap', () => {
let tcp
let listener
const ma = multiaddr('/ip4/127.0.0.1/tcp/9090')
beforeEach((done) => {
tcp = new TCP()
listener = tcp.createListener((conn) => {
pull(conn, conn)
})
listener.on('listening', done)
listener.listen(ma)
})
afterEach((done) => {
listener.close(done)
})
it('simple wrap', (done) => {
const conn = tcp.dial(ma)
conn.setPeerInfo('peerInfo')
const connWrap = new Connection(conn)
pull(
pull.values(['hey']),
connWrap,
pull.collect((err, chunks) => {
expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')])
connWrap.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist()
expect(peerInfo).to.equal('peerInfo')
done()
})
})
)
})
it('buffer wrap', (done) => {
const conn = tcp.dial(ma)
const connWrap = new Connection()
pull(
pull.values(['hey']),
connWrap,
pull.collect((err, chunks) => {
expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')])
done()
})
)
connWrap.setInnerConn(conn)
})
it('overload wrap', (done) => {
const conn = tcp.dial(ma)
const connWrap = new Connection(conn)
connWrap.getPeerInfo = (callback) => {
callback(null, 'none')
}
conn.getPeerInfo((err, peerInfo) => {
expect(err).to.exist()
})
connWrap.getPeerInfo((err, peerInfo) => {
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(chunks).to.be.eql([new Buffer('hey')])
done()
})
)
})
it('matryoshka wrap', (done) => {
const conn = tcp.dial(ma)
const connWrap1 = new Connection(conn)
const connWrap2 = new Connection(connWrap1)
const connWrap3 = new Connection(connWrap2)
conn.getPeerInfo = (callback) => {
callback(null, 'inner doll')
}
pull(
pull.values(['hey']),
connWrap3,
pull.collect((err, chunks) => {
expect(err).to.not.exist()
expect(chunks).to.be.eql([new Buffer('hey')])
connWrap3.getPeerInfo((err, peerInfo) => {
expect(err).to.not.exist()
expect(peerInfo).to.equal('inner doll')
done()
})
})
)
})
})

View File

@ -1,23 +0,0 @@
/* eslint-env mocha */
'use strict'
const tape = require('tape')
const tests = require('interface-transport/tests')
const TCP = require('../src')
// Not adhering to this interface anymore!
describe.skip('interface-transport', () => {
it('works', (done) => {
const common = {
setup (t, cb) {
cb(null, new TCP())
},
teardown (t, cb) {
cb()
}
}
tape.onFinish(done)
tests(tape, common)
})
})