bare-abstract-socket / index.js

// bare-abstract-socket — abstract-namespace AF_UNIX sockets on Linux.
//
// We own only the socket setup (native binding: create the abstract socket, hand
// back fds). Stock, unmodified `bare-pipe` does all stream I/O via `new Pipe(fd)`.
//
// Two surfaces:
//
//   Low level:
//     const { AbstractServer, connect } = require('bare-abstract-socket')
//
//   net-compatible drop-in (so code using #node:net over a UNIX path can switch
//   to an abstract name with a minimal diff):
//     const abstract = require('bare-abstract-socket')
//     const server = abstract.createServer({ allowHalfOpen: false }, onConnection)
//     server.listen('my-name', () => {})
//     const socket = abstract.createConnection('my-name')
//
// The name is the abstract-namespace name WITHOUT the leading NUL — the binding
// adds it. Linux only; abstract sockets are a Linux extension.

const EventEmitter = require('bare-events')
const Pipe = require('bare-pipe')
const binding = require('./binding')

const DEFAULT_BACKLOG = 511

class AbstractServer extends EventEmitter {
  constructor(opts = {}) {
    super()

    this._handle = null
    this._name = null
    // Match bare-pipe's default (true); hyper-svc passes false to avoid the
    // half-open fd leak, so this MUST be honoured on accepted sockets.
    this._allowHalfOpen = opts.allowHalfOpen !== false
    this.listening = false
  }

  listen(name, backlog = DEFAULT_BACKLOG, onListening) {
    if (typeof backlog === 'function') {
      onListening = backlog
      backlog = DEFAULT_BACKLOG
    }
    if (this._handle !== null) throw new Error('Server is already listening')

    // `this` is the ctx; _onConnection is invoked as (err, fd) per accepted conn.
    this._handle = binding.listen(name, backlog, this, this._onConnection)
    this._name = name
    this.listening = true

    if (onListening) this.once('listening', onListening)
    queueMicrotask(() => this.emit('listening'))

    return this
  }

  // Called from native on each accepted connection (and on poll error).
  _onConnection(err, fd) {
    if (err) {
      this.emit('error', new Error('abstract socket poll failed: ' + err))
      return
    }

    // Wrap the accepted fd with stock bare-pipe — it does the stream I/O.
    const socket = new Pipe(fd, { allowHalfOpen: this._allowHalfOpen })
    this.emit('connection', socket)
  }

  address() {
    return this._name
  }

  close(cb) {
    if (this._handle === null) {
      if (cb) queueMicrotask(cb)
      return this
    }

    binding.close(this._handle)
    this._handle = null
    this.listening = false

    if (cb) this.once('close', cb)
    queueMicrotask(() => this.emit('close'))

    return this
  }
}

// Low-level connect: returns a connected bare-pipe stream. Throws synchronously
// if the peer is absent (ECONNREFUSED) or the name is too long.
function connect(name, opts = {}) {
  const fd = binding.connect(name)
  return new Pipe(fd, { allowHalfOpen: opts.allowHalfOpen !== false })
}

// net-compatible surface -----------------------------------------------------

// Mirrors net.createServer(opts?, onConnection?): returns an AbstractServer whose
// listen(name[, cb]) / close([cb]) / 'connection' / 'error' match net's.
function createServer(opts, onConnection) {
  if (typeof opts === 'function') {
    onConnection = opts
    opts = {}
  }
  const server = new AbstractServer(opts || {})
  if (onConnection) server.on('connection', onConnection)
  return server
}

// Mirrors net.createConnection(name[, opts][, onConnect]): NEVER throws
// synchronously — connection errors surface as an async 'error' event, like net.
// On failure a lightweight errored stub is returned so callers can still attach
// 'error' and call destroy().
function createConnection(name, opts, onConnect) {
  if (typeof opts === 'function') {
    onConnect = opts
    opts = {}
  }
  opts = opts || {}

  let fd
  try {
    fd = binding.connect(name)
  } catch (err) {
    const stub = new EventEmitter()
    stub.destroy = () => {}
    queueMicrotask(() => stub.emit('error', err))
    return stub
  }

  const socket = new Pipe(fd, { allowHalfOpen: opts.allowHalfOpen !== false })
  if (onConnect) socket.once('connect', onConnect)
  // bare-pipe opens synchronously for an fd; signal readiness like net does.
  queueMicrotask(() => socket.emit('connect'))
  return socket
}

module.exports = { AbstractServer, connect, createServer, createConnection }