test.js

const test = require('brittle')
const fs = require('bare-fs')
const tcp = require('bare-tcp')
const Pipe = require('.')

const isWindows = Bare.platform === 'win32'
const isLinux = Bare.platform === 'linux' || Bare.platform === 'android'
const ipcHandle = Symbol.for('bare.ipc.handle')

test('server + client', async (t) => {
  t.plan(2)

  const n = name()

  const lc = t.test('lifecycle')
  lc.plan(6)

  const server = Pipe.createServer()
  server
    .on('close', () => t.pass('server closed'))
    .on('connection', (pipe) => {
      pipe
        .on('close', () => lc.pass('server socket closed'))
        .on('data', (data) => lc.alike(data, Buffer.from('hello server')))
        .on('end', () => lc.pass('server ended'))
        .end('hello client')
    })
    .listen(n)

  const client = new Pipe(n)
  client
    .on('close', () => lc.pass('client socket closed'))
    .on('data', (data) => lc.alike(data, Buffer.from('hello client')))
    .on('end', () => lc.pass('client ended'))
    .end('hello server')

  await lc

  server.close()
})

test('server + client, only server writes', async (t) => {
  t.plan(2)

  const n = name()

  const lc = t.test('lifecycle')
  lc.plan(5)

  const server = Pipe.createServer()
  server
    .on('close', () => t.pass('server closed'))
    .on('connection', (pipe) => {
      pipe
        .on('close', () => lc.pass('server socket closed'))
        .on('data', (data) => lc.alike(data, Buffer.from('hello server')))
        .on('end', () => lc.pass('server ended'))
        .end('hello client')
    })
    .listen(n)

  const client = new Pipe(n)
  client
    .on('close', () => lc.pass('client socket closed'))
    .on('data', (data) => lc.alike(data, Buffer.from('hello client')))
    .on('end', () => {
      lc.pass('client ended')
      client.end()
    })

  await lc

  server.close()
})

test('server + client, only client writes', async (t) => {
  t.plan(2)

  const n = name()

  const lc = t.test('lifecycle')
  lc.plan(5)

  const server = Pipe.createServer()
  server
    .on('close', () => t.pass('server closed'))
    .on('connection', (pipe) => {
      pipe
        .on('close', () => lc.pass('server socket closed'))
        .on('data', (data) => lc.alike(data, Buffer.from('hello server')))
        .on('end', () => {
          lc.pass('server ended')
          pipe.end()
        })
    })
    .listen(n)

  const client = new Pipe(n)
  client
    .on('close', () => lc.pass('client socket closed'))
    .on('end', () => lc.pass('client ended'))
    .end('hello server')

  await lc

  server.close()
})

test('socket, read buffer size must be an integer', (t) => {
  t.plan(4)

  t.exception(() => new Pipe({ readBufferSize: 'abc' }), /INVALID_ARGUMENT/)
  t.exception(() => new Pipe({ readBufferSize: 0 }), /INVALID_ARGUMENT/)
  t.exception(() => new Pipe({ readBufferSize: 1.5 }), /INVALID_ARGUMENT/)
  t.exception(() => Pipe.createServer({ readBufferSize: null }), /INVALID_ARGUMENT/)
})

test('socket, read buffer size bounds the chunk size', async (t) => {
  t.plan(2)

  const n = name()
  const payload = Buffer.alloc(64 * 1024, 3)
  const readBufferSize = 1024

  const server = Pipe.createServer({ allowHalfOpen: false }, (pipe) => pipe.end(payload))
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n, { readBufferSize, allowHalfOpen: false })

  const chunks = []
  client.on('data', (data) => chunks.push(data))

  await new Promise((resolve) => client.on('close', resolve))

  t.alike(Buffer.concat(chunks), payload, 'received the whole payload')
  t.ok(
    chunks.every((chunk) => chunk.byteLength <= readBufferSize),
    'no chunk exceeded the read buffer size'
  )

  await new Promise((resolve) => server.close(resolve))
})

test('socket, ipc is coerced to a boolean', async (t) => {
  t.plan(2)

  const socket = new Pipe({ ipc: 1 })

  t.ok(socket.pending, 'constructed with a truthy ipc option')

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))

  const server = Pipe.createServer({ ipc: 0 })

  t.absent(server.listening, 'constructed with a falsy ipc option')

  await new Promise((resolve) => server.close(resolve))
})

test('socket, readyState and pending before connecting', async (t) => {
  t.plan(3)

  const socket = new Pipe()

  t.is(socket.readyState, 'closed', 'closed before connect, as in Node')
  t.ok(socket.pending, 'pending while unconnected')
  t.absent(socket.connecting, 'not connecting until asked to')

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, readyState and pending while connected', async (t) => {
  t.plan(6)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)

  t.ok(client.connecting, 'connecting once asked to')
  t.ok(client.pending, 'pending until connected')

  await new Promise((resolve) => client.on('connect', resolve))

  t.is(client.readyState, 'open', 'open once connected')
  t.absent(client.pending, 'not pending once connected')

  client.destroy()

  await waitFor(client, 'close')

  t.is(client.readyState, 'closed', 'closed once destroyed')
  t.ok(client.pending, 'pending again once the handle is gone, as in Node')

  await new Promise((resolve) => server.close(resolve))
})

test('socket, readyState as the halves end', async (t) => {
  t.plan(4)

  const n = name()

  const server = Pipe.createServer((pipe) => {
    pipe.resume().on('end', () => pipe.end())
  })
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)

  t.is(client.readyState, 'opening', 'opening while connecting')

  await new Promise((resolve) => client.on('connect', resolve))

  t.is(client.readyState, 'open', 'open once connected')

  client.end()

  await new Promise((resolve) => client.on('finish', resolve))

  t.is(client.readyState, 'readOnly', 'read only once the writable half has ended')

  client.destroy()

  await new Promise((resolve) => client.on('close', resolve))

  t.is(client.readyState, 'closed', 'closed once destroyed')

  await new Promise((resolve) => server.close(resolve))
})

test('socket, readyState after the peer ends', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.resume()

  await new Promise((resolve) => client.on('end', resolve))

  t.is(client.readyState, 'writeOnly', 'write only once the peer has ended')

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, readyState once an unconnected pipe is destroyed', async (t) => {
  t.plan(1)

  const socket = new Pipe()

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))

  t.is(socket.readyState, 'closed', 'closed without ever connecting')
})

test('socket, readyState after a failed connect', async (t) => {
  t.plan(2)

  // An empty path is rejected by the underlying handle right away, but the
  // failure is still reported asynchronously, so the pipe stays connecting
  // until it settles, as in Node.
  const socket = new Pipe()
  socket.on('error', () => {})
  socket.connect('')

  t.is(socket.readyState, 'opening', 'still opening while the failure settles')

  await new Promise((resolve) => socket.on('close', resolve))

  t.is(socket.readyState, 'closed', 'closed once the connect has failed')
})

test('socket, readyState while a connect is in flight', async (t) => {
  t.plan(2)

  const socket = new Pipe(name())
  socket.on('error', () => {})

  t.is(socket.readyState, 'opening', 'opening while the connect is in flight')

  await new Promise((resolve) => socket.on('close', resolve))

  t.is(socket.readyState, 'closed', 'closed once the connect has failed')
})

test('socket, connecting is false after failed connect', async (t) => {
  const socket = new Pipe()
  socket.on('error', () => {})
  socket.connect(name())

  await new Promise((resolve) => socket.on('close', resolve))

  t.absent(socket.connecting, 'not connecting after failed connect')
})

test('socket, connect with callback', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe()
  client.connect(n, () => t.pass('connected'))

  await new Promise((resolve) => client.on('connect', resolve))

  t.absent(client.connecting, 'not connecting once connected')

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, connect with options', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end('hello client'))
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe()
  client.connect({ path: n })
  client.on('data', (data) => t.alike(data, Buffer.from('hello client')))

  await new Promise((resolve) => client.on('end', resolve))

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, connect while already connecting', async (t) => {
  t.plan(1)

  const socket = new Pipe(name())
  socket.on('error', () => {})

  t.exception(() => socket.connect(name()), /PIPE_ALREADY_CONNECTED/)

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, connect while already connected', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)

  await new Promise((resolve) => client.on('connect', resolve))

  t.exception(() => client.connect(name()), /PIPE_ALREADY_CONNECTED/)

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, connect with a path that is too long', { skip: isWindows }, async (t) => {
  t.plan(2)

  const socket = new Pipe('/tmp/' + 'a'.repeat(512) + '.sock')
  socket.on('error', (err) => {
    t.is(err.code, 'EINVAL')
    t.absent(socket.connecting, 'not connecting after a rejected path')
  })

  await new Promise((resolve) => socket.on('close', resolve))
})

test(
  'socket, connect with a path that is too long, without an eager open',
  { skip: isWindows },
  async (t) => {
    t.plan(1)

    const socket = new Pipe('/tmp/' + 'a'.repeat(512) + '.sock', { eagerOpen: false })
    socket.on('error', (err) => t.is(err.code, 'EINVAL'))

    await new Promise((resolve) => socket.on('close', resolve))
  }
)

test('socket, connect with a path longer than the maximum', async (t) => {
  t.plan(2)

  const socket = new Pipe()

  t.exception(() => socket.connect('a'.repeat(Pipe.constants.path.MAX_LENGTH + 1)), /INVALID_PATH/)
  t.exception(() => socket.connect(42), /INVALID_PATH/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, connect with a path measured in bytes', async (t) => {
  t.plan(1)

  // Every character takes three bytes, so the byte length is over the limit
  // while the character length is well under it.
  const path = '\u3042'.repeat(Math.floor(Pipe.constants.path.MAX_LENGTH / 3) + 1)

  const socket = new Pipe()

  t.exception(() => socket.connect(path), /INVALID_PATH/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, connect after a rejected path', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe()

  t.exception(() => client.connect(42), /INVALID_PATH/)

  client.connect(n)

  await new Promise((resolve) => client.on('connect', resolve))

  t.absent(client.pending, 'connecting still works after a rejected path')

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, connect failure without an eager open', async (t) => {
  t.plan(1)

  const socket = new Pipe(name(), { eagerOpen: false })
  socket.on('error', (err) => t.ok(err.code, 'connect failed with ' + err.code))

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, connect after a failed connect', async (t) => {
  t.plan(1)

  const socket = new Pipe(name())
  socket.on('error', () => {})

  await new Promise((resolve) => socket.on('close', resolve))

  t.exception(() => socket.connect(name()), /PIPE_IS_CLOSED/)
})

test('socket, connect while a failed connect is still settling', async (t) => {
  t.plan(2)

  // An empty path is the one input the underlying handle rejects synchronously
  // on every platform, though the failure is only reported once it settles.
  const socket = new Pipe()
  socket.on('error', () => {})
  socket.connect('')

  // The pipe is still connecting until the failure settles, as in Node, so
  // another connect is rejected as a duplicate rather than as a closed pipe.
  // Either way it must not be accepted, as the deferred failure would tear it
  // down again.
  t.ok(socket.connecting, 'still connecting while the failure settles')
  t.exception(() => socket.connect(name()), /PIPE_ALREADY_CONNECTED/)

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, open while a failed open is still settling', async (t) => {
  t.plan(1)

  const socket = new Pipe(1 << 24)
  socket.on('error', () => {})

  const [read, write] = Pipe.pipe()

  t.exception(() => socket.open(read), /PIPE_IS_CLOSED/)

  // Neither descriptor was adopted, so close them by hand.
  fs.closeSync(read)
  fs.closeSync(write)

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, connect after being destroyed', async (t) => {
  t.plan(2)

  const socket = new Pipe()
  socket.on('error', () => {})
  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))

  t.exception(() => socket.connect(name()), /PIPE_IS_CLOSED/)
  t.exception(() => socket.open(Pipe.pipe()[0]), /PIPE_IS_CLOSED/)
})

test('socket, connect after being destroyed while connected', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)

  await new Promise((resolve) => client.on('connect', resolve))

  client.destroy()

  await new Promise((resolve) => client.on('close', resolve))

  // Being closed takes precedence over being connected, otherwise the error is
  // misleading.
  t.exception(() => client.connect(n), /PIPE_IS_CLOSED/)

  await new Promise((resolve) => server.close(resolve))
})

test('socket, no connect event once destroyed', async (t) => {
  t.plan(1)

  const [readFd, writeFd] = Pipe.pipe()

  const read = new Pipe(readFd)
  read.on('connect', () => t.fail('connect emitted on a destroyed socket'))
  read.destroy()

  const write = new Pipe(writeFd)
  write.destroy()

  await new Promise((resolve) => read.on('close', resolve))
  await new Promise((resolve) => setTimeout(resolve, 50))

  t.pass('no connect event on a destroyed socket')
})

test('socket, destroy with a connect in flight', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.on('connect', () => t.fail('connect emitted after destroy'))
  client.on('error', () => {})

  // Destroy before the connect has had a chance to complete.
  client.destroy()

  await new Promise((resolve) => client.on('close', resolve))

  t.pass('closed cleanly with a connect in flight')
  t.absent(client.connecting, 'not connecting')

  await new Promise((resolve) => server.close(resolve))
})

test('socket, open with callback and options', async (t) => {
  t.plan(3)

  const [readFd, writeFd] = Pipe.pipe()

  const read = new Pipe()
  read.open(readFd, () => t.pass('read end opened'))

  const write = new Pipe()
  write.open({ fd: writeFd }, () => {
    t.pass('write end opened')
    write.end('hello pipe')
  })

  read.on('data', (data) => t.alike(data, Buffer.from('hello pipe')))

  await new Promise((resolve) => read.on('end', resolve))

  read.destroy()
  write.destroy()
})

test('socket, open with an invalid fd', async (t) => {
  t.plan(1)

  const socket = new Pipe(1 << 24)
  socket.on('error', (err) => t.is(err.code, 'EBADF'))

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, open with an invalid fd type', async (t) => {
  t.plan(4)

  const socket = new Pipe()

  t.exception(() => socket.open('3'), /INVALID_FD/)
  t.exception(() => socket.open({ fd: '3' }), /INVALID_FD/)
  t.exception(() => socket.open(-1), /INVALID_FD/)
  t.exception(() => socket.open(1.5), /INVALID_FD/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, open while already connected', async (t) => {
  t.plan(1)

  const socket = new Pipe(Pipe.pipe()[0])

  t.exception(() => socket.open(Pipe.pipe()[0]), /PIPE_ALREADY_CONNECTED/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, open with an invalid fd, without an eager open', async (t) => {
  t.plan(1)

  const socket = new Pipe(1 << 24, { eagerOpen: false })
  socket.on('error', (err) => t.is(err.code, 'EBADF'))

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, open with an invalid fd after the socket has opened', async (t) => {
  t.plan(1)

  const socket = new Pipe()
  socket.on('error', (err) => t.is(err.code, 'EBADF'))

  // Give the eager open a chance to settle before opening the fd, so that the
  // failure has to be reported against an already opened socket.
  await new Promise((resolve) => setTimeout(resolve, 0))

  socket.open(1 << 24)

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, open after the eager open has settled', async (t) => {
  t.plan(2)

  const [readFd, writeFd] = Pipe.pipe()

  const read = new Pipe()
  const write = new Pipe()

  // Let the eager open settle so that both streams are already waiting on it.
  await new Promise((resolve) => setTimeout(resolve, 0))

  read.open(readFd)
  write.open(writeFd)

  write.end('hello pipe')

  const received = await Promise.race([
    new Promise((resolve) => read.on('data', resolve)),
    new Promise((resolve) => setTimeout(() => resolve(null), 500))
  ])

  t.alike(received, Buffer.from('hello pipe'), 'data flowed after opening')
  t.absent(read.pending, 'no longer pending')

  read.destroy()
  write.destroy()

  await new Promise((resolve) => read.on('close', resolve))
})

test('socket, write with callback', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer((pipe) => {
    pipe.on('data', (data) => {
      t.alike(data, Buffer.from('hello server'))
      pipe.end()
    })
  })
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.write(Buffer.from('hello server'), () => t.pass('write flushed'))

  await new Promise((resolve) => client.on('end', resolve))

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, write with encoding', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer((pipe) => {
    const chunks = []

    pipe
      .on('data', (data) => chunks.push(data))
      .on('end', () => {
        t.alike(Buffer.concat(chunks), Buffer.from('hello serverhello again'))
        pipe.end()
      })
  })
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.write(Buffer.from('hello server').toString('hex'), 'hex', () => t.pass('write flushed'))
  client.write(Buffer.from('hello again').toString('base64'), 'base64')
  client.end()

  await new Promise((resolve) => client.on('end', resolve))

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, write with a handle on a non-ipc pipe', async (t) => {
  t.plan(1)

  const [readFd, writeFd] = Pipe.pipe()

  const read = new Pipe(readFd)
  const write = new Pipe(writeFd)
  const target = new Pipe()

  write.on('error', (err) => {
    t.is(err.code, 'EINVAL')

    read.destroy()
    target.destroy()
  })

  write.write(Buffer.from('here'), target)

  await new Promise((resolve) => write.on('close', resolve))
})

test('socket, write with an invalid handle', async (t) => {
  t.plan(2)

  const socket = new Pipe()

  t.exception(() => socket.write(Buffer.from('here'), {}), /INVALID_IPC_TARGET/)
  t.exception(() => socket.write(Buffer.from('here'), { [ipcHandle]: 42 }), /INVALID_IPC_TARGET/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, write with an invalid handle after destroy', async (t) => {
  t.plan(1)

  const socket = new Pipe()
  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))

  // A destroyed pipe no longer queues handles, but the target is still checked.
  t.exception(() => socket.write(Buffer.from('here'), {}), /INVALID_IPC_TARGET/)
})

test('socket, backpressure', async (t) => {
  t.plan(1)

  const n = name()
  const payload = Buffer.alloc(1024 * 1024, 1)

  const server = Pipe.createServer({ allowHalfOpen: false }, (pipe) => pipe.end(payload))
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n, { readBufferSize: 65536, allowHalfOpen: false })

  const chunks = []
  client.on('data', (data) => chunks.push(data))

  // Stop consuming for a while so that the read buffer fills up and the
  // underlying handle has to be paused.
  client.pause()

  await new Promise((resolve) => setTimeout(resolve, 100))

  client.resume()

  await new Promise((resolve) => client.on('close', resolve))

  t.alike(Buffer.concat(chunks), payload, 'received the whole payload')

  await new Promise((resolve) => server.close(resolve))
})

test('socket, allow half open false', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end('hello client'))
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n, { allowHalfOpen: false })
  client
    .on('end', () => t.pass('client ended'))
    .on('finish', () => t.pass('client finished without being ended'))
    .resume()

  await new Promise((resolve) => client.on('close', resolve))

  await new Promise((resolve) => server.close(resolve))
})

test('socket, allow half open is coerced to a boolean', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end('hello client'))
  server.listen(n)

  await waitForListening(server)

  // A falsy value disables half open, as in Node, so the peer ending also ends
  // the writable half.
  const client = new Pipe(n, { allowHalfOpen: 0 })
  client.on('finish', () => t.pass('client finished without being ended')).resume()

  await new Promise((resolve) => client.on('close', resolve))

  await new Promise((resolve) => server.close(resolve))
})

test('socket, open error is forwarded as a stream error', { skip: isWindows }, async (t) => {
  t.plan(1)

  // A directory can't be watched by the event loop and so can't back a pipe.
  const fd = fs.openSync('/tmp', 'r')

  const socket = new Pipe(fd)
  socket.on('error', (err) => t.is(err.code, 'EINVAL'))
  socket.resume()

  await new Promise((resolve) => socket.on('close', resolve))

  fs.closeSync(fd)
})

test('socket, accept a target without the ipc handle protocol', async (t) => {
  t.plan(1)

  const socket = new Pipe()

  t.exception(() => socket.accept({}), /INVALID_IPC_TARGET/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, accept a target with a forged handle', async (t) => {
  t.plan(2)

  const socket = new Pipe()

  t.exception(() => socket.accept({ [ipcHandle]: 42 }), /INVALID_IPC_TARGET/)
  t.exception(() => socket.accept({ [ipcHandle]: { byteLength: 4096 } }), /INVALID_IPC_TARGET/)

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, ref and unref', async (t) => {
  t.plan(2)

  const socket = new Pipe()

  t.is(socket.unref(), socket, 'unref is chainable')
  t.is(socket.ref(), socket, 'ref is chainable')

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))
})

test('socket, ref and unref after destroy', async (t) => {
  const socket = new Pipe()

  socket.destroy()

  await new Promise((resolve) => socket.on('close', resolve))

  // Redundant during teardown rather than an error, matching the server.
  t.is(socket.ref(), socket)
  t.is(socket.unref(), socket)
})

test('socket, immediate destroy', async (t) => {
  t.plan(2)

  const n = name()

  const lc = t.test('lifecycle')
  lc.plan(3)

  const server = Pipe.createServer()
  server
    .on('close', () => t.pass('server closed'))
    .on('connection', (pipe) => {
      pipe
        .on('close', () => lc.pass('server socket closed'))
        .on('end', () => {
          lc.pass('server ended')
          pipe.end()
        })
    })
    .listen(n)

  const client = new Pipe(n)
  client
    .on('close', () => lc.pass('client socket closed'))
    .on('end', () => lc.pass('client ended'))
    .destroy()

  await lc

  server.close()
})

test('socket, remote credentials on both ends', async (t) => {
  const n = name()

  let accepted = null
  const server = Pipe.createServer()
  const connection = new Promise((resolve) => {
    server.on('connection', (pipe) => {
      accepted = pipe
      resolve(pipe.remoteCredentials())
    })
  })
  server.listen(n)

  const client = new Pipe(n)
  await new Promise((resolve) => client.on('connect', resolve))

  const seenByServer = await connection
  const seenByClient = client.remoteCredentials()

  t.is(seenByServer.pid, Bare.pid, 'the server sees the client process')
  t.is(seenByClient.pid, Bare.pid, 'the client sees the server process')

  if (isWindows) {
    t.ok(/^S-1-5-/.test(seenByServer.sid), 'the server sees a user SID')
    t.ok(
      seenByClient.sid === seenByServer.sid || seenByClient.sid === 'S-1-5-32-544',
      'the client sees the owner of the pipe, the same user unless elevated'
    )
    t.is(seenByServer.uid, null, 'no uid on Windows')
    t.is(seenByServer.gid, null, 'no gid on Windows')
  } else {
    t.ok(Number.isInteger(seenByServer.uid) && seenByServer.uid >= 0, 'the server sees a uid')
    t.ok(Number.isInteger(seenByServer.gid) && seenByServer.gid >= 0, 'the server sees a gid')
    t.alike(seenByClient, seenByServer, 'both ends see the same account')
    t.is(seenByServer.sid, null, 'no SID on Unix')
  }

  client.destroy()
  accepted.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, remote credentials before connecting', (t) => {
  const pipe = new Pipe()

  t.exception(() => pipe.remoteCredentials(), /PIPE_NOT_CONNECTED/)

  pipe.destroy()
})

test('socket, remote credentials after a failed connect', async (t) => {
  const pipe = new Pipe(name())

  await new Promise((resolve) => pipe.on('error', resolve))

  t.exception(() => pipe.remoteCredentials(), /PIPE_IS_CLOSED/)
})

test('socket, remote credentials of a pipe that is not a socket', { skip: isWindows }, (t) => {
  const [read, write] = Pipe.pipe()

  const reader = new Pipe(read)
  const writer = new Pipe(write)

  try {
    reader.remoteCredentials()
    t.fail('a pipe that is not a socket has no peer')
  } catch (err) {
    t.is(err.code, 'ENOTSOCK')
  }

  reader.destroy()
  writer.destroy()
})

test('socket, remote credentials of an anonymous pipe', { skip: !isWindows }, (t) => {
  const [read, write] = Pipe.pipe()

  const reader = new Pipe(read)
  const writer = new Pipe(write)

  t.is(reader.remoteCredentials().pid, Bare.pid, 'the server end sees the client process')
  t.is(writer.remoteCredentials().pid, Bare.pid, 'the client end sees the server process')
  t.is(reader.remoteCredentials().sid, null, 'the server end cannot tell when it connected')
  t.ok(/^S-1-5-/.test(writer.remoteCredentials().sid), 'the client end sees the owner of the pipe')

  reader.destroy()
  writer.destroy()
})

test('socket, remote credentials after destroy', async (t) => {
  const n = name()

  const server = Pipe.createServer()
  server.on('connection', (pipe) => pipe.destroy())
  server.listen(n)

  const client = new Pipe(n)
  await new Promise((resolve) => client.on('connect', resolve))

  client.destroy()

  t.exception(() => client.remoteCredentials(), /PIPE_IS_CLOSED/)

  await new Promise((resolve) => server.close(resolve))
})

test('server, connection listener as the only argument', async (t) => {
  t.plan(2)

  const n = name()

  const server = new Pipe.Server((pipe) => {
    t.pass('server got a connection')
    pipe.end()
  })
  server.listen(n, () => t.pass('listening'))

  const client = new Pipe(n)

  await new Promise((resolve) => client.on('end', resolve))

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('server, connection listener that throws', async (t) => {
  t.plan(3)

  const n = name()

  const server = Pipe.createServer()

  server.on('error', () => t.fail('a listener bug must not become a server error'))

  // Capture the connection before the listener that throws runs, as listeners
  // are called in the order they were added.
  const accepted = new Promise((resolve) => server.on('connection', resolve))

  server.on('connection', () => {
    throw new Error('listener bug')
  })

  // The exception propagates out of the accept, as in Node, so it has to be
  // caught here to keep it from taking the test process down with it.
  const onuncaught = (err) => t.is(err.message, 'listener bug', 'the exception propagates')

  Bare.on('uncaughtException', onuncaught)

  t.teardown(() => Bare.off('uncaughtException', onuncaught))

  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n).on('error', noop)

  const pipe = await accepted

  t.absent(pipe.destroying, 'the connection handed to the listener survives')

  pipe.end('hello')

  const data = await Promise.race([
    new Promise((resolve) => client.on('data', resolve)),
    new Promise((resolve) => setTimeout(() => resolve(null), 500))
  ])

  t.alike(data, Buffer.from('hello'), 'and is still usable')

  client.destroy()
  server.close()
})

test('server, pause on connect', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer({ pauseOnConnect: true }, (pipe) => {
    pipe.on('data', (data) => {
      t.alike(data, Buffer.from('hello server'))
      pipe.end()
    })
  })
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.end('hello server')

  await new Promise((resolve) => client.on('end', resolve))

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('server, closing', async (t) => {
  t.plan(3)

  const server = Pipe.createServer()

  t.absent(server.closing, 'not closing before listening')

  server.listen(name())

  await waitForListening(server)

  server.close()

  t.ok(server.closing, 'closing as soon as close is called')

  await new Promise((resolve) => server.on('close', resolve))

  t.absent(server.closing, 'no longer closing once closed')
})

test('server, listening is false after close', async (t) => {
  t.plan(3)

  const server = Pipe.createServer()
  server.listen(name())

  await waitForListening(server)

  t.ok(server.listening, 'listening while bound')

  server.close(() => {
    t.absent(server.listening, 'not listening after close')
    t.is(server.address(), null, 'no address after close')
  })
})

test('server, address while listening', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer()
  server.listen(n)

  await waitForListening(server)

  t.ok(server.listening, 'listening')
  t.is(server.address(), n, 'address is the bound path')

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen with backlog and callback', async (t) => {
  t.plan(1)

  const server = Pipe.createServer()
  server.listen(name(), 128, () => t.pass('listening'))

  await waitForListening(server)

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen with options', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer()
  server.listen({ path: n, backlog: 128 })

  await waitForListening(server)

  t.is(server.address(), n, 'address is the bound path')

  const other = name()

  const withoutBacklog = Pipe.createServer()
  withoutBacklog.listen({ path: other })

  await waitForListening(withoutBacklog)

  t.is(withoutBacklog.address(), other, 'address is the bound path')

  await new Promise((resolve) => withoutBacklog.close(resolve))
  await new Promise((resolve) => server.close(resolve))
})

test('server, listen with an invalid backlog', async (t) => {
  t.plan(6)

  const server = Pipe.createServer()

  t.exception(() => server.listen(name(), 'abc'), /INVALID_ARGUMENT/)
  t.exception(() => server.listen({ path: name(), backlog: 'abc' }), /INVALID_ARGUMENT/)

  // A falsy backlog is rejected rather than quietly replaced by the default,
  // which only stands in for an absent one.
  t.exception(() => server.listen({ path: name(), backlog: NaN }), /INVALID_ARGUMENT/)
  t.exception(() => server.listen({ path: name(), backlog: false }), /INVALID_ARGUMENT/)
  t.exception(() => server.listen({ path: name(), backlog: '' }), /INVALID_ARGUMENT/)

  // A rejected argument must not leave the server wedged.
  const n = name()
  server.listen(n)

  await waitForListening(server)

  t.is(server.address(), n, 'listening still works after a rejected backlog')

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen with a zero backlog', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end())
  server.listen(n, 0)

  await waitForListening(server)

  const client = new Pipe(n)
  client.resume()

  await new Promise((resolve) => client.on('end', resolve))

  t.pass('a zero backlog falls back to the default')

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen with a path longer than the maximum', async (t) => {
  t.plan(2)

  const server = Pipe.createServer()

  t.exception(() => server.listen('a'.repeat(Pipe.constants.path.MAX_LENGTH + 1)), /INVALID_PATH/)
  t.exception(() => server.listen({ path: 42 }), /INVALID_PATH/)

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen while already listening', async (t) => {
  t.plan(1)

  const server = Pipe.createServer()
  server.listen(name())

  t.exception(() => server.listen(name()), /SERVER_ALREADY_LISTENING/)

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen after close', async (t) => {
  t.plan(3)

  const server = Pipe.createServer()
  server.listen(name())

  await waitForListening(server)
  await new Promise((resolve) => server.close(resolve))

  t.absent(server.closing, 'no longer closing once closed')

  // A fully closed server is reusable, as in Node.
  const n = name()
  server.listen(n)

  await waitForListening(server)

  t.ok(server.listening, 'listening again')
  t.is(server.address(), n, 'bound to the new path')

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen while closing', async (t) => {
  t.plan(1)

  const server = Pipe.createServer()
  server.listen(name())

  await waitForListening(server)

  server.close()

  t.exception(() => server.listen(name()), /SERVER_IS_CLOSED/)

  await new Promise((resolve) => server.on('close', resolve))
})

test('server, listen while connections drain', async (t) => {
  t.plan(2)

  const n = name()

  const server = Pipe.createServer()
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  const pipe = await new Promise((resolve) => server.on('connection', resolve))

  server.close()

  t.ok(server.closing, 'closing while the connection drains')
  t.exception(() => server.listen(name()), /SERVER_IS_CLOSED/)

  pipe.destroy()
  client.destroy()

  await new Promise((resolve) => server.on('close', resolve))
})

test('server, listen on the same path after closing', async (t) => {
  t.plan(2)

  const n = name()

  const first = Pipe.createServer()
  first.listen(n)

  await waitForListening(first)
  await new Promise((resolve) => first.close(resolve))

  // The path is released on close, so it can be bound again.
  const second = Pipe.createServer()
  second.listen(n)

  await waitForListening(second)

  t.ok(second.listening, 'bound the released path')
  t.is(second.address(), n, 'address is the released path')

  await new Promise((resolve) => second.close(resolve))
})

test('server, no listening event once closed', async (t) => {
  t.plan(1)

  const server = Pipe.createServer()
  server.on('listening', () => t.fail('listening emitted on a closing server'))
  server.listen(name())
  server.close()

  await new Promise((resolve) => server.on('close', resolve))
  await new Promise((resolve) => setTimeout(resolve, 50))

  t.pass('no listening event on a closed server')
})

test('server, bind error', async (t) => {
  t.plan(3)

  const n = name()

  const first = Pipe.createServer()
  first.listen(n)

  await waitForListening(first)

  const second = Pipe.createServer()
  second.on('error', (err) => {
    t.is(err.code, 'EADDRINUSE')
    t.absent(second.listening, 'not listening after a failed bind')
    t.is(second.address(), null, 'no address after a failed bind')
  })
  second.listen(n)

  await new Promise((resolve) => first.close(resolve))
})

test('server, listen again after a failed bind', async (t) => {
  t.plan(2)

  const n = name()

  const first = Pipe.createServer()
  first.listen(n)

  await waitForListening(first)

  const second = Pipe.createServer()
  second.listen(n)

  const err = await new Promise((resolve) => second.on('error', resolve))

  t.is(err.code, 'EADDRINUSE', 'bind failed')

  const other = name()
  second.listen(other)

  await waitForListening(second)

  t.is(second.address(), other, 'bound to another path after the failure')

  await new Promise((resolve) => second.close(resolve))
  await new Promise((resolve) => first.close(resolve))
})

test('server, close without listening', async (t) => {
  t.plan(2)

  const server = Pipe.createServer()

  await new Promise((resolve) => server.close(resolve))

  t.pass('closed without ever listening')

  await new Promise((resolve) => server.close(resolve))

  t.pass('closed again')
})

test('server, close after closing', async (t) => {
  t.plan(3)

  const server = Pipe.createServer()
  server.listen(name())

  await waitForListening(server)
  await new Promise((resolve) => server.close(resolve))

  t.is(server.close(), server, 'close is chainable once closed')

  await new Promise((resolve) => server.close(resolve))

  t.pass('close called back again once closed')
  t.is(server.listenerCount('close'), 0, 'no close listeners left behind')
})

test('server, close after listening again', async (t) => {
  t.plan(2)

  const server = Pipe.createServer()
  server.listen(name())

  await waitForListening(server)
  await new Promise((resolve) => server.close(resolve))

  server.listen(name())

  await waitForListening(server)

  let closed = 0
  server.on('close', () => closed++)

  await new Promise((resolve) => server.close(resolve))

  t.is(closed, 1, 'the second close closed the server')
  t.absent(server.listening, 'not listening')
})

test('server, close callbacks while closing', async (t) => {
  t.plan(2)

  const server = Pipe.createServer()
  server.listen(name())

  await waitForListening(server)

  let closes = 0
  server.on('close', () => closes++)

  let callbacks = 0
  server.close(() => callbacks++)
  server.close(() => callbacks++)
  server.close(() => callbacks++)

  await new Promise((resolve) => setTimeout(resolve, 50))

  t.is(callbacks, 3, 'every close callback ran')
  t.is(closes, 1, 'closed once')
})

test('server, close after a failed bind', async (t) => {
  t.plan(2)

  const n = name()

  const first = Pipe.createServer()
  first.listen(n)

  await waitForListening(first)

  const events = []

  const second = Pipe.createServer()
  second.on('error', (err) => events.push(err.code))
  second.on('close', () => events.push('close'))
  second.listen(n)
  second.close(() => events.push('callback'))

  await new Promise((resolve) => setTimeout(resolve, 50))

  t.alike(events, ['EADDRINUSE', 'close', 'callback'], 'reported the error before closing')
  t.absent(second.listening, 'not listening')

  await new Promise((resolve) => first.close(resolve))
})

test('server, close releases the path right away', { skip: isWindows }, async (t) => {
  t.plan(4)

  const n = name()

  let connections = 0

  const server = Pipe.createServer((pipe) => {
    connections++

    // Close while this connection is still open, then try to connect again.
    server.close()

    t.absent(server.listening, 'not listening as soon as close is called')
    t.is(server.address(), null, 'no address as soon as close is called')

    const late = new Pipe(n)
    late.on('error', () => {
      t.pass('a later connection is refused')
      pipe.destroy()
    })
  })
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.on('error', () => {})

  await new Promise((resolve) => server.on('close', resolve))

  t.is(connections, 1, 'only the connection made while listening was accepted')

  client.destroy()
})

test('server, close with a connect in flight', async (t) => {
  t.plan(1)

  const n = name()

  const server = Pipe.createServer(() => t.fail('connection emitted after close'))
  server.listen(n)

  await waitForListening(server)

  const client = new Pipe(n)
  client.on('error', () => {})

  // Close before the pending connect has been accepted.
  server.close()

  await new Promise((resolve) => server.on('close', resolve))

  t.pass('closed cleanly with a connect in flight')

  client.destroy()
})

test('server, ref and unref', async (t) => {
  t.plan(3)

  const server = Pipe.createServer()

  t.is(server.unref(), server, 'unref is chainable without a handle')

  server.listen(name(), () => t.pass('listening'))

  t.is(server.ref(), server, 'ref is chainable with a handle')

  server.unref()
  server.ref()

  await waitForListening(server)

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen owner only', async (t) => {
  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end('hello'))
  server.listen({ path: n, ownerOnly: true })

  await waitForListening(server)

  if (!isWindows) t.is(fs.statSync(n).mode & 0o777, 0o600, 'the socket is private to its owner')

  const client = new Pipe(n)
  const data = await new Promise((resolve, reject) => {
    client.on('data', resolve).on('error', reject)
  })

  t.alike(data, Buffer.from('hello'), 'the owner can still connect')

  if (isWindows) {
    t.ok(/^S-1-5-/.test(client.remoteCredentials().sid), 'the server runs as a user SID')
  }

  client.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen owner only with a backlog', { skip: isWindows }, async (t) => {
  const n = name()

  const server = Pipe.createServer()
  server.listen(n, 16, { ownerOnly: true })

  await waitForListening(server)

  t.is(fs.statSync(n).mode & 0o777, 0o600)

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen with null options', async (t) => {
  const server = Pipe.createServer()
  server.listen(name(), 511, null)

  await waitForListening(server)

  t.ok(server.listening)

  await new Promise((resolve) => server.close(resolve))
})

test('server, listen owner only that is not a boolean', (t) => {
  const server = Pipe.createServer()

  t.exception(() => server.listen({ path: name(), ownerOnly: 'false' }), /INVALID_ARGUMENT/)
  t.absent(server.listening)
})

test('server, listen owner only on an abstract socket', { skip: !isLinux }, (t) => {
  const server = Pipe.createServer()

  t.exception(
    () => server.listen({ path: '\0' + name().slice(5), ownerOnly: true }),
    /INVALID_ARGUMENT/,
    'an abstract socket has no mode to restrict'
  )
  t.absent(server.listening)
})

test('createConnection, arguments', async (t) => {
  t.plan(5)

  const n = name()

  const server = Pipe.createServer((pipe) => pipe.end('hello client'))
  server.listen(n)

  await waitForListening(server)

  const a = Pipe.createConnection(n)
  const b = Pipe.createConnection(n, () => t.pass('connected with callback'))
  const c = Pipe.createConnection({ path: n })
  const d = Pipe.createConnection(n, { allowHalfOpen: false })

  for (const socket of [a, b, c, d]) {
    const chunks = []

    socket.on('data', (data) => chunks.push(data))

    await new Promise((resolve) => socket.on('end', resolve))

    t.alike(Buffer.concat(chunks), Buffer.from('hello client'))

    socket.destroy()
  }

  await new Promise((resolve) => server.close(resolve))
})

test('pipe, data flow between the ends', (t) => {
  t.plan(1)

  const fds = Pipe.pipe()

  const read = new Pipe(fds[0])

  read.on('data', (data) => t.alike(data, Buffer.from('hello pipe')))

  const write = new Pipe(fds[1])

  write.end('hello pipe')
})

test('pipe, readyState of both ends', async (t) => {
  t.plan(3)

  const [readFd, writeFd] = Pipe.pipe()

  const read = new Pipe(readFd)
  const write = new Pipe(writeFd)

  t.is(read.readyState, 'readOnly', 'read end is read only')
  t.is(write.readyState, 'writeOnly', 'write end is write only')

  read.on('data', (data) => t.alike(data, Buffer.from('hello pipe')))

  write.end('hello pipe')

  await new Promise((resolve) => read.on('end', resolve))

  read.destroy()
  write.destroy()
})

test('ipc, data only', { skip: isWindows }, (t) => {
  t.plan(1)

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  right.on('data', (data) => {
    t.alike(data, Buffer.from('hello ipc'))
    left.destroy()
    right.destroy()
  })

  left.write(Buffer.from('hello ipc'))
})

test('ipc, pipe handle pass', { skip: isWindows }, (t) => {
  t.plan(3)

  const echo = name()

  const server = Pipe.createServer()
  server.on('connection', (peer) => peer.pipe(peer)).listen(echo)

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  right
    .on('handle', (type) => {
      t.is(type, Pipe.constants.handle.NAMED_PIPE)

      const received = new Pipe()
      right.accept(received)

      received.on('data', (data) => {
        t.alike(data, Buffer.from('ping'))
        received.destroy()
        left.destroy()
        right.destroy()
        server.close()
      })
      received.write('ping')
    })
    .resume()

  const peer = new Pipe(echo)
  peer.on('connect', () => {
    left.write(Buffer.from('here'), peer, () => {
      t.pass('handle sent')
      peer.destroy()
    })
  })
})

test('ipc, remote credentials of a received pipe', { skip: isWindows }, (t) => {
  t.plan(3)

  const echo = name()

  const server = Pipe.createServer()
  server.listen(echo)

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  right
    .on('handle', () => {
      const received = right.accept(new Pipe())
      const credentials = received.remoteCredentials()

      t.ok(Number.isInteger(credentials.uid), 'the received pipe still reports its peer')
      t.is(credentials.sid, null, 'no SID for a received pipe')

      received.destroy()
      left.destroy()
      right.destroy()
      server.close()
    })
    .resume()

  const peer = new Pipe(echo)
  peer.on('connect', () => {
    left.write(Buffer.from('here'), peer, () => {
      t.pass('handle sent')
      peer.destroy()
    })
  })
})

test('ipc, tcp handle pass', { skip: isWindows }, async (t) => {
  t.plan(3)

  const server = tcp.createServer()
  server.on('connection', (sock) => sock.pipe(sock)).listen()

  await waitForListening(server)

  const { port } = server.address()

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  const peer = tcp.createConnection(port)

  right
    .on('handle', (type) => {
      t.is(type, Pipe.constants.handle.TCP)

      const received = new tcp.Socket()
      right.accept(received)

      received
        .on('data', (data) => {
          t.alike(data, Buffer.from('ping'))
          received.destroy()
          peer.destroy()
          left.destroy()
          right.destroy()
          server.close()
        })
        .write('ping')
    })
    .resume()

  peer.on('connect', () => {
    left.write(Buffer.from('here'), peer, () => t.pass('handle sent'))
  })
})

test('ipc, multiple pending handles drain in order', { skip: isWindows }, async (t) => {
  t.plan(6)

  const echoA = name()
  const echoB = name()

  const serverA = Pipe.createServer()
  serverA.on('connection', (peer) => peer.pipe(peer)).listen(echoA)

  const serverB = Pipe.createServer()
  serverB.on('connection', (peer) => peer.pipe(peer)).listen(echoB)

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  const received = []

  right
    .on('handle', (type) => {
      const target = type === Pipe.constants.handle.NAMED_PIPE ? new Pipe() : new tcp.Socket()
      right.accept(target)
      received.push({ type, target })

      if (received.length === 3) {
        t.is(received[0].type, Pipe.constants.handle.NAMED_PIPE, 'first is pipe')
        t.is(received[1].type, Pipe.constants.handle.TCP, 'second is tcp')
        t.is(received[2].type, Pipe.constants.handle.NAMED_PIPE, 'third is pipe')

        for (const { target } of received) target.destroy()
        peerA.destroy()
        peerT.destroy()
        peerB.destroy()
        left.destroy()
        right.destroy()
        serverA.close()
        serverB.close()
        server.close()
      }
    })
    .resume()

  const server = tcp.createServer()
  server.on('connection', (sock) => sock.pipe(sock)).listen()
  await waitForListening(server)
  const { port } = server.address()

  const peerA = new Pipe(echoA)
  const peerT = tcp.createConnection(port)
  const peerB = new Pipe(echoB)

  let connected = 0
  const tryWrite = () => {
    if (++connected < 3) return

    left.write(Buffer.from('a'), peerA, () => t.pass('first sent'))
    left.write(Buffer.from('b'), peerT, () => t.pass('second sent'))
    left.write(Buffer.from('c'), peerB, () => t.pass('third sent'))
  }

  peerA.on('connect', tryWrite)
  peerT.on('connect', tryWrite)
  peerB.on('connect', tryWrite)
})

test('ipc, each handle rides with its own message', { skip: isWindows }, async (t) => {
  t.plan(5)

  const echo = name()

  const server = Pipe.createServer((peer) => peer.pipe(peer))
  server.listen(echo)

  const tcpServer = tcp.createServer((sock) => sock.pipe(sock))
  tcpServer.listen()

  await waitForListening(server)
  await waitForListening(tcpServer)

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  const types = []
  const chunks = []

  // The data seen so far when each handle arrived. Ancillary data forces a read
  // boundary, but where the boundary falls is platform specific: Linux ends the
  // read after the message that carries the handle, coalescing everything
  // written before it, whereas macOS ends the read before it. The prefix
  // therefore only bounds which message carried the handle from above.
  const prefixes = []

  right
    .on('handle', (type) => {
      types.push(type)
      prefixes.push(Buffer.concat(chunks).toString())

      right.accept(type === Pipe.constants.handle.TCP ? new tcp.Socket() : new Pipe()).destroy()
    })
    .on('data', (data) => chunks.push(data))

  const peer = new Pipe(echo)
  const peerTcp = tcp.createConnection(tcpServer.address().port)

  await new Promise((resolve) => peer.on('connect', resolve))
  await new Promise((resolve) => peerTcp.on('connect', resolve))

  // Queued in one tick, so a single batch is split into segments around the
  // two handles.
  left.write(Buffer.from('one'))
  left.write(Buffer.from('two'), peer)
  left.write(Buffer.from('three'))
  left.write(Buffer.from('four'), peerTcp)
  left.write(Buffer.from('five'))

  await new Promise((resolve) => setTimeout(resolve, 100))

  t.alike(
    types,
    [Pipe.constants.handle.NAMED_PIPE, Pipe.constants.handle.TCP],
    'received both handles in order'
  )
  t.ok('onetwo'.startsWith(prefixes[0]), 'the pipe arrived no later than the second message')
  t.ok(
    'onetwothreefour'.startsWith(prefixes[1]),
    'the socket arrived no later than the fourth message'
  )
  // Two handles on the same message would be surfaced from the same read, and
  // so would see the same prefix.
  t.not(prefixes[0], prefixes[1], 'the handles arrived on separate reads')
  t.alike(Buffer.concat(chunks), Buffer.from('onetwothreefourfive'), 'received all data in order')

  peer.destroy()
  peerTcp.destroy()
  left.destroy()
  right.destroy()

  await new Promise((resolve) => server.close(resolve))
  await new Promise((resolve) => tcpServer.close(resolve))
})

test('ipc, handle and data writes in the same batch', { skip: isWindows }, async (t) => {
  t.plan(2)

  const echo = name()

  const server = Pipe.createServer({ allowHalfOpen: false }, (peer) => peer.pipe(peer))
  server.listen(echo)

  await waitForListening(server)

  const [a, b] = tcp.socketpair()

  const left = new Pipe(a, { ipc: true })
  const right = new Pipe(b, { ipc: true })

  const types = []
  const received = []

  right
    .on('handle', (type) => {
      types.push(type)
      right.accept(new Pipe()).destroy()
    })
    .on('data', (data) => received.push(data))

  const peer = new Pipe(echo)

  await new Promise((resolve) => peer.on('connect', resolve))

  left.write(Buffer.from('one'))
  left.write(Buffer.from('two'), peer)
  left.write(Buffer.from('three'))
  left.end(Buffer.from('four'))

  await new Promise((resolve) => setTimeout(resolve, 100))

  t.alike(types, [Pipe.constants.handle.NAMED_PIPE], 'received a single handle')
  t.alike(Buffer.concat(received), Buffer.from('onetwothreefour'), 'received all data in order')

  peer.destroy()
  left.destroy()
  right.destroy()

  await new Promise((resolve) => server.close(resolve))
})

test('socket, coerces a view that is not a buffer', async (t) => {
  t.plan(1)

  const [read, write] = Pipe.pipe()

  const reader = new Pipe(read)
  const writer = new Pipe(write)

  const chunks = []
  reader.on('data', (chunk) => chunks.push(chunk))

  const buffer = new ArrayBuffer(4)
  new Uint8Array(buffer).set([1, 2, 3, 4])

  // A view that is not a buffer shares its memory with the buffer it is coerced
  // to, so the bytes have to arrive unchanged, offset and length included.
  writer.end(new DataView(buffer, 1, 2))

  await waitFor(reader, 'end')

  t.alike(Buffer.concat(chunks), Buffer.from([2, 3]))

  reader.destroy()
})

test('socket, rejects a chunk that is not a view', async (t) => {
  t.plan(1)

  const [read, write] = Pipe.pipe()

  const reader = new Pipe(read)
  const writer = new Pipe(write)

  reader.resume()

  // An array buffer is not a view, so there is nothing for the write request to
  // point at. It has to fail the write rather than reach the binding.
  writer.write(new ArrayBuffer(4))

  const err = await new Promise((resolve) => writer.on('error', resolve))

  t.is(err.code, 'INVALID_ARGUMENT')

  reader.destroy()
})

function noop() {}

function waitForListening(server) {
  if (server.listening) return Promise.resolve()

  return waitFor(server, 'listening')
}

function waitFor(emitter, event) {
  return new Promise((resolve, reject) => {
    emitter.on(event, done).on('error', done)

    function done(err) {
      emitter.off(event, done).off('error', done)

      err ? reject(err) : resolve()
    }
  })
}

function name() {
  const name =
    'bare-pipe-' + Math.random().toString(16).slice(2) + Math.random().toString(16).slice(2)
  return isWindows ? '\\\\.\\pipe\\' + name : '/tmp/' + name + '.sock'
}