autobase-doctor / node_modules / bare-tcp / index.js

const EventEmitter = require('bare-events')
const { Duplex } = require('bare-stream')
const dns = require('bare-dns')
const binding = require('./binding')
const constants = require('./lib/constants')
const errors = require('./lib/errors')
const ip = require('./lib/ip')

const defaultReadBufferSize = 65536
const empty = Buffer.alloc(0)

exports.Socket = class TCPSocket extends Duplex {
  constructor(opts = {}) {
    const { readBufferSize = defaultReadBufferSize, allowHalfOpen = true, eagerOpen = true } = opts

    super({ eagerOpen })

    this._state = 0

    this._allowHalfOpen = allowHalfOpen

    this._keepAlive = 0
    this._keepAliveInitialDelay = 0
    this._noDelay = 0

    this._localAddress = null
    this._remoteAddress = null

    this._pendingOpen = null
    this._pendingWrite = null
    this._pendingFinal = null
    this._pendingDestroy = null

    this._timer = null
    this._timeout = 0

    this._buffer = Buffer.alloc(readBufferSize)

    this._addresses = null
    this._errors = null

    this._handle = binding.init(
      this._buffer,
      this,
      noop,
      this._onconnect,
      this._onreset,
      this._onread,
      this._onwrite,
      this._onfinal,
      this._onclose
    )
  }

  get connecting() {
    return (this._state & constants.state.CONNECTING) !== 0
  }

  get pending() {
    return (this._state & constants.state.CONNECTED) === 0
  }

  get timeout() {
    return this._timeout || undefined // For Node.js compatibility
  }

  get readyState() {
    if (this._state & constants.state.CONNECTED) {
      return 'open'
    }

    return 'opening'
  }

  get localAddress() {
    if (this._localAddress) return this._localAddress.address
  }

  get localFamily() {
    if (this._localAddress) return `IPv${this._localAddress.family}`
  }

  get localPort() {
    if (this._localAddress) return this._localAddress.port
  }

  get remoteAddress() {
    if (this._remoteAddress) return this._remoteAddress.address
  }

  get remoteFamily() {
    if (this._remoteAddress) return `IPv${this._remoteAddress.family}`
  }

  get remotePort() {
    if (this._remoteAddress) return this._remoteAddress.port
  }

  connect(port, host = 'localhost', opts = {}, onconnect) {
    if (this._state & constants.state.CONNECTING || this._state & constants.state.CONNECTED) {
      throw errors.SOCKET_ALREADY_CONNECTED('Socket is already connected')
    }

    this._state |= constants.state.CONNECTING

    if (typeof host === 'function') {
      onconnect = host
      host = 'localhost'
    } else if (typeof opts === 'function') {
      onconnect = opts
      opts = {}
    }

    let family = 0

    if (typeof port === 'object' && port !== null) {
      opts = port || {}
      port = opts.port || 0
      host = opts.host || 'localhost'
      family = opts.family || 0
    }

    if (!host) host = 'localhost'

    const {
      lookup = dns.lookup,
      hints,
      keepAlive = false,
      keepAliveInitialDelay = 0,
      noDelay = false,
      timeout
    } = opts

    const type = ip.isIP(host)

    if (type === 0) {
      lookup(host, { all: true, family, hints }, (err, addresses) => {
        if (this._state & constants.state.CLOSING) return

        this._state &= ~constants.state.CONNECTING

        if (err || addresses.length === 0) {
          if (!err) {
            err = new Error(`No address found for host "${host}"`)
            err.code = 'ENOTFOUND'
          }

          this.emit('lookup', err, null, 0, host)

          if (this._pendingOpen) this._continueOpen(err)
          else this.destroy(err)
          return
        }

        for (const { address, family } of addresses) {
          this.emit('lookup', null, address, family, host)
        }

        const [{ address, family }, ...rest] = addresses

        if (rest.length > 0) {
          this._addresses = rest.map(({ address, family }) => [
            port,
            address,
            { ...opts, family },
            onconnect
          ])

          this._errors = []
        }

        this.connect(port, address, { ...opts, family }, onconnect)
      })

      return this
    }

    family = type

    try {
      binding.connect(this._handle, port, host, family)

      if (keepAlive) {
        this._keepAlive = keepAlive
        this._keepAliveInitialDelay = keepAliveInitialDelay
      }
      if (noDelay) this._noDelay = noDelay
      if (timeout) this.setTimeout(timeout)

      if (onconnect) this.once('connect', onconnect)
    } catch (err) {
      queueMicrotask(() => {
        if (this._pendingOpen) this._pendingOpen(err)
        else this.destroy(err)
      })
    }

    return this
  }

  setKeepAlive(enable = false, delay = 0) {
    if (typeof enable === 'number') {
      delay = enable
      enable = false
    }

    delay = Math.floor(delay / 1000)

    if (delay === 0) enable = false

    binding.keepalive(this._handle, enable, delay)

    return this
  }

  setNoDelay(enable = true) {
    binding.nodelay(this._handle, enable)

    return this
  }

  setTimeout(ms, ontimeout) {
    if (ms === 0) {
      clearTimeout(this._timer)

      this._timer = null
    } else {
      if (ontimeout) this.once('timeout', ontimeout)

      this._timer = setTimeout(() => this.emit('timeout'), ms)
      this._timer.unref()
    }

    this._timeout = ms

    return this
  }

  ref() {
    binding.ref(this._handle)

    return this
  }

  unref() {
    binding.unref(this._handle)

    return this
  }

  _open(cb) {
    if (this._state & constants.state.CONNECTED) return cb(null)

    this._pendingOpen = cb
  }

  _read() {
    if ((this._state & constants.state.READING) === 0) {
      this._state |= constants.state.READING

      binding.resume(this._handle)
    }
  }

  _writev(batch, cb) {
    this._pendingWrite = [cb, batch]

    binding.writev(
      this._handle,
      batch.map(({ chunk }) => chunk)
    )
  }

  _final(cb) {
    this._pendingFinal = cb

    binding.end(this._handle)
  }

  _predestroy() {
    if (this._state & constants.state.CLOSING) return
    this._state |= constants.state.CLOSING

    binding.close(this._handle)
  }

  _destroy(err, cb) {
    if (this._state & constants.state.CLOSING) return cb(err)
    this._state |= constants.state.CLOSING

    this._pendingDestroy = cb

    binding.close(this._handle)
  }

  _continueOpen(err) {
    if (this._pendingOpen === null) return
    const cb = this._pendingOpen
    this._pendingOpen = null
    cb(err)
  }

  _continueWrite(err) {
    if (this._pendingWrite === null) return
    const cb = this._pendingWrite[0]
    this._pendingWrite = null
    cb(err)
  }

  _continueFinal(err) {
    if (this._pendingFinal === null) return
    const cb = this._pendingFinal
    this._pendingFinal = null
    cb(err)
  }

  _continueDestroy() {
    if (this._pendingDestroy === null) return
    const cb = this._pendingDestroy
    this._pendingDestroy = null
    cb(null)
  }

  _reset() {
    this._state = 0

    this._localAddress = null
    this._remoteAddress = null

    binding.reset(this._handle)
  }

  _onconnect(err) {
    if (err) {
      if (this._addresses !== null) {
        this._errors.push(err)

        if (this._addresses.length > 0) return this._reset()

        err = this._errors.length === 1 ? this._errors[0] : new AggregateError(this._errors)
      }

      if (this._pendingOpen) this._continueOpen(err)
      else this.destroy(err)
      return
    }

    if (this._keepAlive) this.setKeepAlive(this._keepAlive, this._keepAliveInitialDelay)
    if (this._noDelay) this.setNoDelay()

    this._localAddress = binding.address(this._handle, true)
    this._remoteAddress = binding.address(this._handle, false)

    this._state |= constants.state.CONNECTED
    this._state &= ~constants.state.CONNECTING
    this._continueOpen()

    this.emit('connect')
  }

  _onreset(err) {
    if (err) {
      this._errors.push(err)
      this.destroy(this._errors.length === 1 ? this._errors[0] : new AggregateError(this._errors))
      return
    }

    this.connect(...this._addresses.shift())
  }

  _onread(err, read) {
    if (this._timer) this._timer.refresh()

    if (err) {
      this.destroy(err)
      return
    }

    if (read === 0) {
      this.push(null)
      if (this._allowHalfOpen === false) this.end()
      return
    }

    const copy = Buffer.allocUnsafe(read)
    copy.set(this._buffer.subarray(0, read))

    if (this.push(copy) === false && this.destroying === false) {
      this._state &= ~constants.state.READING

      binding.pause(this._handle)
    }
  }

  _onwrite(err) {
    if (this._timer) this._timer.refresh()

    this._continueWrite(err)
  }

  _onfinal(err) {
    this._continueFinal(err)
  }

  _onclose() {
    clearTimeout(this._timer)

    this._continueDestroy()
  }
}

exports.Server = class TCPServer extends EventEmitter {
  constructor(opts = {}, onconnection) {
    if (typeof opts === 'function') {
      onconnection = opts
      opts = {}
    }

    super()

    const {
      readBufferSize = defaultReadBufferSize,
      allowHalfOpen = true,
      keepAlive = false,
      keepAliveInitialDelay = 0,
      noDelay = false,
      pauseOnConnect = false
    } = opts

    this._state = 0

    this._readBufferSize = readBufferSize
    this._allowHalfOpen = allowHalfOpen
    this._keepAlive = keepAlive
    this._keepAliveInitialDelay = keepAliveInitialDelay
    this._noDelay = noDelay
    this._pauseOnConnect = pauseOnConnect

    this._address = null
    this._connections = new Set()

    this._error = null
    this._handle = null

    if (onconnection) this.on('connection', onconnection)
  }

  get listening() {
    return (this._state & constants.state.BOUND) !== 0
  }

  get closing() {
    return (this._state & constants.state.CLOSING) !== 0
  }

  get connections() {
    return this._connections
  }

  address() {
    if ((this._state & constants.state.BOUND) === 0) return null

    const { address, family, port } = this._address

    return { address, family: `IPv${family}`, port }
  }

  listen(port = 0, host = 'localhost', backlog = 511, opts = {}, onlistening) {
    if (this._state & constants.state.BINDING || this._state & constants.state.BOUND) {
      throw errors.SERVER_ALREADY_LISTENING('Server is already listening')
    }

    if (this._state & constants.state.CLOSING) {
      throw errors.SERVER_IS_CLOSED('Server is closed')
    }

    this._state |= constants.state.BINDING

    if (typeof port === 'function') {
      onlistening = port
      port = 0
    } else if (typeof host === 'function') {
      onlistening = host
      host = 'localhost'
    } else if (typeof backlog === 'function') {
      onlistening = backlog
      backlog = 511
    } else if (typeof opts === 'function') {
      onlistening = opts
      opts = {}
    }

    let family = 0

    if (typeof port === 'object' && port !== null) {
      opts = port || {}
      port = opts.port || 0
      host = opts.host || 'localhost'
      family = opts.family || 0
      backlog = opts.backlog || 511
    }

    if (!host) host = 'localhost'
    if (!backlog) backlog = 511

    const { lookup = dns.lookup, hints } = opts

    const type = ip.isIP(host)

    if (type === 0) {
      lookup(host, { family, hints }, (err, address, family) => {
        if (this._state & constants.state.CLOSING) return

        this.emit('lookup', err, address, family, host)

        this._state &= ~constants.state.BINDING

        if (err) return this.emit('error', err)

        this.listen(port, address, backlog, { ...opts, family }, onlistening)
      })

      return this
    }

    family = type

    this._handle = binding.init(
      empty,
      this,
      this._onconnection,
      noop,
      noop,
      noop,
      noop,
      noop,
      this._onclose
    )

    if (this._state & constants.state.UNREFED) binding.unref(this._handle)

    try {
      binding.bind(this._handle, port, host, backlog, family)

      this._address = binding.address(this._handle, true)

      this._state |= constants.state.BOUND
      this._state &= ~constants.state.BINDING

      if (onlistening) this.once('listening', onlistening)

      queueMicrotask(() => this.emit('listening'))
    } catch (err) {
      this._error = err

      binding.close(this._handle)
    }

    return this
  }

  close(onclose) {
    if (onclose) this.once('close', onclose)

    if (this._state & constants.state.CLOSING) return this
    this._state |= constants.state.CLOSING

    this._closeMaybe()

    return this
  }

  ref() {
    this._state &= ~constants.state.UNREFED

    if (this._handle !== null) binding.ref(this._handle)

    return this
  }

  unref() {
    this._state |= constants.state.UNREFED

    if (this._handle !== null) binding.unref(this._handle)

    return this
  }

  _closeMaybe() {
    if (this._state & constants.state.CLOSING && this._connections.size === 0) {
      if (this._handle !== null) binding.close(this._handle)
      else queueMicrotask(() => this.emit('close'))
    }
  }

  _onconnection(err) {
    if (err) {
      this.emit('error', err)
      return
    }

    if (this._state & constants.state.CLOSING) return

    const socket = new exports.Socket({
      readBufferSize: this._readBufferSize,
      allowHalfOpen: this._allowHalfOpen,
      eagerOpen: !this._pauseOnConnect
    })

    try {
      binding.accept(this._handle, socket._handle)

      socket._localAddress = binding.address(socket._handle, true)
      socket._remoteAddress = binding.address(socket._handle, false)

      socket._state |= constants.state.CONNECTED

      this._connections.add(socket)

      if (this._keepAlive) socket.setKeepAlive(this._keepAlive, this._keepAliveInitialDelay)
      if (this._noDelay) socket.setNoDelay()

      socket.on('close', () => {
        this._connections.delete(socket)
        this._closeMaybe()
      })

      this.emit('connection', socket)
    } catch (err) {
      socket.destroy()

      this.emit('error', err)
    }
  }

  _onclose() {
    const err = this._error

    this._state &= ~constants.state.BINDING
    this._error = null
    this._handle = null
    this._address = null

    if (err) this.emit('error', err)
    else this.emit('close')
  }
}

exports.constants = constants
exports.errors = errors

exports.isIP = ip.isIP
exports.isIPv4 = ip.isIPv4
exports.isIPv6 = ip.isIPv6

exports.createConnection = function createConnection(port, host, opts, onconnect) {
  if (typeof host === 'function') {
    onconnect = host
    host = 'localhost'
  } else if (typeof opts === 'function') {
    onconnect = opts
    opts = {}
  }

  if (typeof port === 'object' && port !== null) {
    opts = port || {}
    port = opts.port || 0
    host = opts.host || 'localhost'
  }

  return new exports.Socket(opts).connect(port, host, opts, onconnect)
}

// For Node.js compatibility
exports.connect = exports.createConnection

exports.createServer = function createServer(opts, onconnection) {
  return new exports.Server(opts, onconnection)
}

function noop() {}