|
|
|
@ -1,7 +1,6 @@ |
|
|
|
const FIFO = require('fast-fifo') |
|
|
|
const sodium = require('sodium-universal') |
|
|
|
const c = require('compact-encoding') |
|
|
|
const bind = require('bind-easy') |
|
|
|
const b4a = require('b4a') |
|
|
|
const peer = require('./peer') |
|
|
|
const errors = require('./errors') |
|
|
|
@ -13,14 +12,16 @@ const TMP = b4a.alloc(32) |
|
|
|
const EMPTY_ARRAY = [] |
|
|
|
|
|
|
|
module.exports = class IO { |
|
|
|
constructor (table, { maxWindow = 80, bind = 0, firewalled = true, onrequest, onresponse = noop, ontimeout = noop } = {}) { |
|
|
|
constructor (table, udx, { maxWindow = 80, port = 0, firewalled = true, onrequest, onresponse = noop, ontimeout = noop } = {}) { |
|
|
|
this.table = table |
|
|
|
this.udx = udx |
|
|
|
this.inflight = [] |
|
|
|
this.clientSocket = null |
|
|
|
this.serverSocket = null |
|
|
|
this.firewalled = firewalled !== false |
|
|
|
this.ephemeral = true |
|
|
|
this.congestion = new CongestionWindow(maxWindow) |
|
|
|
this.networkInterfaces = udx.watchNetworkInterfaces() |
|
|
|
|
|
|
|
this.onrequest = onrequest |
|
|
|
this.onresponse = onresponse |
|
|
|
@ -33,13 +34,13 @@ module.exports = class IO { |
|
|
|
this._drainInterval = null |
|
|
|
this._destroying = null |
|
|
|
this._binding = null |
|
|
|
this._bind = bind |
|
|
|
this._port = port |
|
|
|
} |
|
|
|
|
|
|
|
onmessage (socket, buffer, rinfo) { |
|
|
|
if (buffer.byteLength < 2 || !(rinfo.port > 0 && rinfo.port < 65536)) return |
|
|
|
onmessage (socket, buffer, { host, port }) { |
|
|
|
if (buffer.byteLength < 2 || !(port > 0 && port < 65536)) return |
|
|
|
|
|
|
|
const from = { id: null, host: rinfo.address, port: rinfo.port } |
|
|
|
const from = { id: null, host, port } |
|
|
|
const state = { start: 1, end: buffer.byteLength, buffer } |
|
|
|
const expectedSocket = this.firewalled ? this.clientSocket : this.serverSocket |
|
|
|
const external = socket !== expectedSocket |
|
|
|
@ -114,16 +115,11 @@ module.exports = class IO { |
|
|
|
req.onerror(errors.createDestroyedError(), req) |
|
|
|
} |
|
|
|
|
|
|
|
this._destroying = new Promise((resolve) => { |
|
|
|
let missing = 2 |
|
|
|
|
|
|
|
this.serverSocket.close(done) |
|
|
|
this.clientSocket.close(done) |
|
|
|
|
|
|
|
function done () { |
|
|
|
if (--missing === 0) resolve() |
|
|
|
} |
|
|
|
}) |
|
|
|
this._destroying = Promise.allSettled([ |
|
|
|
this.serverSocket.close(), |
|
|
|
this.clientSocket.close(), |
|
|
|
this.networkInterfaces.destroy() |
|
|
|
]) |
|
|
|
|
|
|
|
return this._destroying |
|
|
|
} |
|
|
|
@ -135,19 +131,32 @@ module.exports = class IO { |
|
|
|
} |
|
|
|
|
|
|
|
async _bindSockets () { |
|
|
|
const serverSocket = typeof this._bind === 'function' ? await this._bind() : await bind.udp(this._bind) |
|
|
|
const serverSocket = this.udx.createSocket() |
|
|
|
|
|
|
|
try { |
|
|
|
serverSocket.bind(this._port) |
|
|
|
} catch { |
|
|
|
try { |
|
|
|
serverSocket.bind() |
|
|
|
} catch (err) { |
|
|
|
await serverSocket.close() |
|
|
|
throw err |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
const clientSocket = this.udx.createSocket() |
|
|
|
|
|
|
|
try { |
|
|
|
// TODO: we should reroll the socket is it's close to our preferred range of ports
|
|
|
|
// to avoid it being accidentally opened
|
|
|
|
// We'll prop need additional APIs for that
|
|
|
|
this.clientSocket = await bind.udp() |
|
|
|
this.serverSocket = serverSocket |
|
|
|
clientSocket.bind() |
|
|
|
} catch (err) { |
|
|
|
await new Promise((resolve) => serverSocket.close(resolve)) |
|
|
|
await serverSocket.close() |
|
|
|
await clientSocket.close() |
|
|
|
throw err |
|
|
|
} |
|
|
|
|
|
|
|
this.clientSocket = clientSocket |
|
|
|
this.serverSocket = serverSocket |
|
|
|
|
|
|
|
this.serverSocket.on('message', this.onmessage.bind(this, this.serverSocket)) |
|
|
|
this.clientSocket.on('message', this.onmessage.bind(this, this.clientSocket)) |
|
|
|
|
|
|
|
@ -242,8 +251,7 @@ class Request { |
|
|
|
reply (value, opts = {}) { |
|
|
|
const socket = opts.socket || this.socket |
|
|
|
const to = opts.to || this.from |
|
|
|
const onflush = opts.onflush || null |
|
|
|
this._sendReply(0, value || null, opts.token !== false, opts.closerNodes !== false, to, socket, onflush) |
|
|
|
this._sendReply(0, value || null, opts.token !== false, opts.closerNodes !== false, to, socket) |
|
|
|
} |
|
|
|
|
|
|
|
error (code, opts = {}) { |
|
|
|
@ -255,7 +263,7 @@ class Request { |
|
|
|
relay (value, to, opts) { |
|
|
|
const socket = (opts && opts.socket) || this.socket |
|
|
|
const buffer = this._encodeRequest(null, value, to, socket) |
|
|
|
socket.send(buffer, 0, buffer.byteLength, to.port, to.host) |
|
|
|
socket.trySend(buffer, to.port, to.host) |
|
|
|
} |
|
|
|
|
|
|
|
send (force = false) { |
|
|
|
@ -280,7 +288,7 @@ class Request { |
|
|
|
if (this.destroyed) return |
|
|
|
this.sent++ |
|
|
|
this._io.congestion.send() |
|
|
|
this.socket.send(this._buffer, 0, this._buffer.byteLength, this.to.port, this.to.host) |
|
|
|
this.socket.trySend(this._buffer, this.to.port, this.to.host) |
|
|
|
if (this._timeout) clearTimeout(this._timeout) |
|
|
|
this._timeout = setTimeout(oncycle, 1000, this) |
|
|
|
} |
|
|
|
@ -298,7 +306,7 @@ class Request { |
|
|
|
this.onerror(err || errors.createDestroyedError(), this) |
|
|
|
} |
|
|
|
|
|
|
|
_sendReply (error, value, token, hasCloserNodes, from, socket, onflush) { |
|
|
|
_sendReply (error, value, token, hasCloserNodes, from, socket) { |
|
|
|
if (socket === null || this.destroyed) return |
|
|
|
|
|
|
|
const id = this._io.ephemeral === false && socket === this._io.serverSocket |
|
|
|
@ -324,7 +332,7 @@ class Request { |
|
|
|
if (error > 0) c.uint.encode(state, error) |
|
|
|
if (value) c.buffer.encode(state, value) |
|
|
|
|
|
|
|
socket.send(state.buffer, 0, state.buffer.byteLength, from.port, from.host, onflush) |
|
|
|
socket.trySend(state.buffer, from.port, from.host) |
|
|
|
} |
|
|
|
|
|
|
|
_encodeRequest (token, value, to, socket) { |
|
|
|
|
Kasper Isager Dalsgarð
2 years ago
GitHub