Socket
Socket
Sign inDemoInstall

libp2p-tcp

Package Overview
Dependencies
33
Maintainers
2
Versions
57
Alerts
File Explorer

Advanced tools

npm Scripts

Install Socket

Detect and block malicious and high-risk dependencies

Install

Comparing version 0.13.1 to 0.14.0

src/constants.js
src/ip-port-to-multiaddr.js
src/socket-to-conn.js

6

.vscode/launch.json

@@ -13,9 +13,7 @@ {

"args": [
"-u",
"tdd",
"--timeout",
"999999",
"--colors",
// "--grep",
// "getAddrs",
"--grep",
"close listener with connections, through timeout",
"${workspaceFolder}/test/compliance.spec.js"

@@ -22,0 +20,0 @@ ],

15

CHANGELOG.md

@@ -0,1 +1,16 @@

<a name="0.14.0"></a>
# [0.14.0](https://github.com/libp2p/js-libp2p-tcp/compare/v0.13.1...v0.14.0) (2019-09-16)
### Features
* change api to async / await ([#112](https://github.com/libp2p/js-libp2p-tcp/issues/112)) ([cf7d1b8](https://github.com/libp2p/js-libp2p-tcp/commit/cf7d1b8))
### BREAKING CHANGES
* All places in the API that used callbacks are now replaced with async/await. The API has also been updated according to the latest `interface-transport` version, https://github.com/libp2p/interface-transport/tree/v0.6.0#api.
<a name="0.13.1"></a>

@@ -2,0 +17,0 @@ ## [0.13.1](https://github.com/libp2p/js-libp2p-tcp/compare/v0.13.0...v0.13.1) (2019-08-08)

25

package.json
{
"name": "libp2p-tcp",
"version": "0.13.1",
"version": "0.14.0",
"description": "Node.js implementation of the TCP module that libp2p uses, which implements the interface-connection and interface-transport interfaces",

@@ -14,8 +14,6 @@ "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",

"release-major": "aegir-release -t node --type major --no-build",
"coverage": "aegir coverage",
"coverage-publish": "aegir coverage --provider coveralls"
"coverage": "nyc --reporter=text --reporter=lcov npm run test:node"
},
"pre-push": [
"lint",
"test"
"lint"
],

@@ -42,18 +40,17 @@ "repository": {

"dirty-chai": "^2.0.1",
"interface-transport": "~0.3.6",
"pull-stream": "^3.6.14"
"interface-transport": "^0.6.1",
"sinon": "^7.3.1"
},
"dependencies": {
"abortable-iterator": "^2.1.0",
"class-is": "^1.1.0",
"debug": "^4.1.1",
"interface-connection": "~0.3.3",
"err-code": "^2.0.0",
"ip-address": "^6.1.0",
"lodash.includes": "^4.3.0",
"lodash.isfunction": "^3.0.9",
"mafmt": "^6.0.7",
"multiaddr": "^6.1.0",
"once": "^1.4.0",
"stream-to-pull-stream": "^1.7.3"
"mafmt": "^6.0.9",
"multiaddr": "^7.1.0",
"stream-to-it": "^0.1.1"
},
"contributors": [
"Alan Shaw <alan.shaw@protocol.ai>",
"Alan Shaw <alan@tableflip.io>",

@@ -60,0 +57,0 @@ "David Dias <daviddias.p@gmail.com>",

40

README.md

@@ -16,3 +16,3 @@ # js-libp2p-tcp

> JavaScript implementation of the TCP module for libp2p. It exposes the [interface-transport](https://github.com/libp2p/interface-connection) for dial/listen. `libp2p-tcp` is a very thin shim that adds support for dialing to a `multiaddr`. This small shim will enable libp2p to use other different transports.
> JavaScript implementation of the TCP module for libp2p. It exposes the [interface-transport](https://github.com/libp2p/interface-connection) for dial/listen. `libp2p-tcp` is a very thin shim that adds support for dialing to a `multiaddr`. This small shim will enable libp2p to use other transports.

@@ -45,5 +45,6 @@ ## Lead Maintainer

const multiaddr = require('multiaddr')
const pull = require('pull-stream')
const pipe = require('it-pipe')
const { collect } = require('streaming-iterables')
const mh = multiaddr('/ip4/127.0.0.1/tcp/9090')
const addr = multiaddr('/ip4/127.0.0.1/tcp/9090')

@@ -54,4 +55,4 @@ const tcp = new TCP()

console.log('new connection opened')
pull(
pull.values(['hello']),
pipe(
['hello'],
socket

@@ -61,19 +62,14 @@ )

listener.listen(mh, () => {
console.log('listening')
await listener.listen(addr)
console.log('listening')
pull(
tcp.dial(mh),
pull.collect((err, values) => {
if (!err) {
console.log(`Value: ${values.toString()}`)
} else {
console.log(`Error: ${err}`)
}
const socket = await tcp.dial(addr)
const values = await pipe(
socket,
collect
)
console.log(`Value: ${values.toString()}`)
// Close connection after reading
listener.close()
}),
)
})
// Close connection after reading
await listener.close()
```

@@ -95,3 +91,3 @@

`libp2p-tcp` accepts TCP addresses both IPFS and non IPFS encapsulated addresses, i.e:
`libp2p-tcp` accepts TCP addresses as both IPFS and non IPFS encapsulated addresses, i.e:

@@ -101,3 +97,3 @@ `/ip4/127.0.0.1/tcp/4001`

Both for dialing and listening.
(both for dialing and listening)

@@ -104,0 +100,0 @@ ### Connection

152

src/index.js
'use strict'
const net = require('net')
const toPull = require('stream-to-pull-stream')
const mafmt = require('mafmt')
const withIs = require('class-is')
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 errCode = require('err-code')
const log = require('debug')('libp2p:tcp')
const toConnection = require('./socket-to-conn')
const createListener = require('./listener')
const { AbortError } = require('abortable-iterator')
const { CODE_CIRCUIT, CODE_P2P } = require('./constants')
const assert = require('assert')
function noop () {}
/**
* @class TCP
*/
class TCP {
/**
* @constructor
* @param {object} options
* @param {Upgrader} options.upgrader
*/
constructor ({ upgrader }) {
assert(upgrader, 'An upgrader must be provided. See https://github.com/libp2p/interface-transport#upgrader.')
this._upgrader = upgrader
}
class TCP {
dial (ma, options, callback) {
if (isFunction(options)) {
callback = options
options = {}
/**
* @async
* @param {Multiaddr} ma
* @param {object} options
* @param {AbortSignal} options.signal Used to abort dial requests
* @returns {Connection} An upgraded Connection
*/
async dial (ma, options) {
options = options || {}
const socket = await this._connect(ma, options)
const maConn = toConnection(socket, { remoteAddr: ma, signal: options.signal })
log('new outbound connection %s', maConn.remoteAddr)
const conn = await this._upgrader.upgradeOutbound(maConn)
log('outbound connection %s upgraded', maConn.remoteAddr)
return conn
}
/**
* @private
* @param {Multiaddr} ma
* @param {object} options
* @param {AbortSignal} options.signal Used to abort dial requests
* @returns {Promise<Socket>} Resolves a TCP Socket
*/
_connect (ma, options = {}) {
if (options.signal && options.signal.aborted) {
throw new AbortError()
}
callback = once(callback || noop)
return new Promise((resolve, reject) => {
const start = Date.now()
const cOpts = ma.toOptions()
const cOpts = ma.toOptions()
log('Connecting to %s %s', cOpts.port, cOpts.host)
log('dialing %s:%s', cOpts.host, cOpts.port)
const rawSocket = net.connect(cOpts)
const rawSocket = net.connect(cOpts)
const onError = err => {
err.message = `connection error ${cOpts.host}:${cOpts.port}: ${err.message}`
done(err)
}
rawSocket.once('timeout', () => {
log('timeout')
rawSocket.emit('error', new Error('Timeout'))
})
const onTimeout = () => {
log('connnection timeout %s:%s', cOpts.host, cOpts.port)
const err = errCode(new Error(`connection timeout after ${Date.now() - start}ms`), 'ERR_CONNECT_TIMEOUT')
// Note: this will result in onError() being called
rawSocket.emit('error', err)
}
rawSocket.once('error', callback)
const onConnect = () => {
log('connection opened %s:%s', cOpts.host, cOpts.port)
done()
}
rawSocket.once('connect', () => {
rawSocket.removeListener('error', callback)
callback()
})
const onAbort = () => {
log('connection aborted %s:%s', cOpts.host, cOpts.port)
rawSocket.destroy()
done(new AbortError())
}
const socket = toPull.duplex(rawSocket)
const done = err => {
rawSocket.removeListener('error', onError)
rawSocket.removeListener('timeout', onTimeout)
rawSocket.removeListener('connect', onConnect)
options.signal && options.signal.removeEventListener('abort', onAbort)
const conn = new Connection(socket)
if (err) return reject(err)
resolve(rawSocket)
}
conn.getObservedAddrs = (callback) => {
return callback(null, [ma])
}
return conn
rawSocket.on('error', onError)
rawSocket.on('timeout', onTimeout)
rawSocket.on('connect', onConnect)
options.signal && options.signal.addEventListener('abort', onAbort)
})
}
/**
* Creates a TCP listener. The provided `handler` function will be called
* anytime a new incoming Connection has been successfully upgraded via
* `upgrader.upgradeInbound`.
* @param {*} [options]
* @param {function(Connection)} handler
* @returns {Listener} A TCP listener
*/
createListener (options, handler) {
if (isFunction(options)) {
if (typeof options === 'function') {
handler = options
options = {}
}
handler = handler || noop
return createListener(handler)
options = options || {}
return createListener({ handler, upgrader: this._upgrader }, options)
}
/**
* Takes a list of `Multiaddr`s and returns only valid TCP addresses
* @param {Multiaddr[]} multiaddrs
* @returns {Multiaddr[]} Valid TCP multiaddrs
*/
filter (multiaddrs) {
if (!Array.isArray(multiaddrs)) {
multiaddrs = [multiaddrs]
}
multiaddrs = Array.isArray(multiaddrs) ? multiaddrs : [multiaddrs]
return multiaddrs.filter((ma) => {
if (includes(ma.protoNames(), 'p2p-circuit')) {
return multiaddrs.filter(ma => {
if (ma.protoCodes().includes(CODE_CIRCUIT)) {
return false
}
if (includes(ma.protoNames(), 'ipfs')) {
ma = ma.decapsulate('ipfs')
}
return mafmt.TCP.matches(ma)
return mafmt.TCP.matches(ma.decapsulateCode(CODE_P2P))
})

@@ -82,0 +136,0 @@ }

186

src/listener.js
'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 EventEmitter = require('events')
const log = require('debug')('libp2p:tcp:listener')
const toConnection = require('./socket-to-conn')
const { CODE_P2P } = require('./constants')
const ProtoFamily = { ip4: 'IPv4', ip6: 'IPv6' }
const getMultiaddr = require('./get-multiaddr')
const IPFS_CODE = 421
const CLOSE_TIMEOUT = 2000
function noop () {}
module.exports = (handler) => {
module.exports = ({ handler, upgrader }, options) => {
const listener = new EventEmitter()
const server = net.createServer((socket) => {
// Avoid uncaught errors cause by unstable connections
socket.on('error', noop)
const server = net.createServer(async socket => {
// Avoid uncaught errors caused by unstable connections
socket.on('error', err => log('socket error', err))
const addr = getMultiaddr(socket)
if (!addr) {
if (socket.remoteAddress === undefined) {
log('connection closed before p2p connection made')
} else {
log('error interpreting incoming p2p connection')
}
return
}
const maConn = toConnection(socket)
log('new inbound connection %s', maConn.remoteAddr)
log('new connection', addr.toString())
const conn = await upgrader.upgradeInbound(maConn)
log('inbound connection %s upgraded', maConn.remoteAddr)
const s = toPull.duplex(socket)
trackConn(server, maConn)
s.getObservedAddrs = (cb) => {
cb(null, [addr])
}
trackSocket(server, socket)
const conn = new Connection(s)
handler(conn)
if (handler) 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'))
server
.on('listening', () => listener.emit('listening'))
.on('error', err => listener.emit('error', err))
.on('close', () => listener.emit('close'))
// Keep track of open connections to destroy in case of timeout
server.__connections = {}
server.__connections = []
listener.close = (options, callback) => {
if (typeof options === 'function') {
callback = options
options = {}
}
callback = callback || noop
options = options || {}
listener.close = () => {
if (!server.listening) return
const timeout = setTimeout(() => {
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)
server.close(callback)
server.once('close', () => {
clearTimeout(timeout)
return new Promise((resolve, reject) => {
server.__connections.forEach(maConn => maConn.close())
server.close(err => err ? reject(err) : resolve())
})
}
let ipfsId
let listeningAddr
let peerId, listeningAddr
listener.listen = (ma, callback) => {
listener.listen = ma => {
listeningAddr = ma
if (includes(ma.protoNames(), 'ipfs')) {
ipfsId = getIpfsId(ma)
listeningAddr = ma.decapsulate('ipfs')
peerId = ma.getPeerId()
if (peerId) {
listeningAddr = ma.decapsulateCode(CODE_P2P)
}
const lOpts = listeningAddr.toOptions()
log('Listening on %s %s', lOpts.port, lOpts.host)
return server.listen(lOpts.port, lOpts.host, callback)
return new Promise((resolve, reject) => {
const { host, port } = listeningAddr.toOptions()
server.listen(port, host, err => {
if (err) return reject(err)
log('Listening on %s %s', port, host)
resolve()
})
})
}
listener.getAddrs = (callback) => {
const multiaddrs = []
listener.getAddrs = () => {
let addrs = []
const address = server.address()
if (!address) {
return callback(new Error('Listener is not ready yet'))
throw new Error('Listener is not ready yet')
}

@@ -107,33 +78,9 @@

// 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 (listeningAddr.toString().startsWith('/ip4')) {
addrs = addrs.concat(getMulitaddrs('ip4', address.address, address.port))
} else if (address.family === 'IPv6') {
addrs = addrs.concat(getMulitaddrs('ip6', address.address, address.port))
}
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 addrs.map(ma => peerId ? ma.encapsulate(`/p2p/${peerId}`) : ma)
}

@@ -144,15 +91,34 @@

function getIpfsId (ma) {
return ma.stringTuples().filter((tuple) => {
return tuple[0] === IPFS_CODE
})[0][1]
function getMulitaddrs (proto, ip, port) {
const toMa = ip => multiaddr(`/${proto}/${ip}/tcp/${port}`)
return (isAnyAddr(ip) ? getNetworkAddrs(ProtoFamily[proto]) : [ip]).map(toMa)
}
function trackSocket (server, socket) {
const key = `${socket.remoteAddress}:${socket.remotePort}`
server.__connections[key] = socket
function isAnyAddr (ip) {
return ['0.0.0.0', '::'].includes(ip)
}
socket.on('close', () => {
delete server.__connections[key]
})
/**
* @private
* @param {string} family One of ['IPv6', 'IPv4']
* @returns {string[]} an array of ip address strings
*/
function getNetworkAddrs (family) {
return Object.values(os.networkInterfaces()).reduce((addresses, netAddrs) => {
netAddrs.forEach(netAddr => {
// Add the ip of each matching network interface
if (netAddr.family === family) addresses.push(netAddr.address)
})
return addresses
}, [])
}
function trackConn (server, maConn) {
server.__connections.push(maConn)
const untrackConn = () => {
server.__connections = server.__connections.filter(c => c !== maConn)
}
maConn.conn.once('close', untrackConn)
}
dist/index.js
dist/index.js.map
dist/index.min.js
dist/index.min.js.map
src/get-multiaddr.js
.travis.yml

Sorry, the diff of this file is not supported yet

docs/index.html

Sorry, the diff of this file is not supported yet

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap

About

Packages

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc