bare-pipe

bare-pipe — Native I/O pipes for JavaScript. It is a native addon and requires Bare >=1.16.0.

npm i bare-pipe

Usage

const Pipe = require('bare-pipe')

const stdout = new Pipe(1)

stdout.write('Hello world!\n')

API

Pipe

new Pipe(path: string | number, opts?: PipeOptions)

Create a new pipe. If path is a number, it is treated as a file descriptor to open. If it is a string, it is treated as a path to connect to.

Overloads:

new Pipe(path: string | number, opts?: PipeOptions)
new Pipe(opts?: PipeOptions)

Parameters

_destroy(err: Error | null, cb: StreamCallback): void

Parameters

_final(cb: StreamCallback): void

Parameters

_open(cb: StreamCallback): void

Parameters

_predestroy(): void

_read(size: number): void

Parameters

_write(data: unknown, encoding: StreamEncoding, cb: StreamCallback): void

Parameters

_writev(batch: { chunk: unknown; encoding: StreamEncoding }[], cb: StreamCallback): void

Parameters

accept<T extends IPCAcceptable>(target: T): T

Accept a pending handle into target. target must implement the IPCAcceptable protocol. Call this synchronously from the 'handle' event listener.

Parameters

Returns T — target, for chaining the accepted handle into an expression.

Throws

• INVALID_IPC_TARGET — target does not implement the IPC handle protocol.

closed: boolean

connect(path: string, opts?: PipeConnectOptions, onconnect?: () => void): this

Connect the pipe to path. onconnect is called when the connection is established.

Overloads:

connect(path: string, opts?: PipeConnectOptions, onconnect?: () => void): this
connect(path: string, onconnect: () => void): this
connect(opts: PipeConnectOptions, onconnect?: () => void): this

Parameters

Throws

• PIPE_ALREADY_CONNECTED — the pipe is already connecting or connected.

connecting: boolean

Whether the pipe is currently connecting.

cork(): void

destroy(err?: Error | null): void

Parameters

destroyed: boolean

destroying: boolean

end(cb?: StreamCallback): this

Overloads:

end(cb?: StreamCallback): this
end(data: unknown, encoding?: BufferEncoding, cb?: StreamCallback): this
end(data: unknown, cb?: StreamCallback): this

Parameters

errored: Error | null

open(fd: number, opts?: PipeOpenOptions, onconnect?: () => void): this

Open the pipe on the given file descriptor.

Overloads:

open(fd: number, opts?: PipeOpenOptions, onconnect?: () => void): this
open(fd: number, onconnect: () => void): this
open(opts: PipeOpenOptions & { fd: number }, onconnect?: () => void): this

Parameters

pause(): this

pending: boolean

Whether the pipe has not yet connected.

Pipe.constants

Pipe.constants: {
  state: {
    CONNECTING: number
    CONNECTED: number
    BINDING: number
    BOUND: number
    READING: number
    CLOSING: number
    CLOSED: number
    UNREFED: number
    READABLE: number
    WRITABLE: number
  }
  path: {
    MAX_LENGTH: number
  }
  handle: {
    NAMED_PIPE: number
    TCP: number
    UDP: number
  }
}

Object containing internal state constants and handle type constants.

Pipe.createConnection

Pipe.createConnection(path: string, opts?: CreateConnectionOptions, onconnect?: () => void): Pipe

Create a new pipe and connect it to path. Shorthand for new Pipe(options).connect(path, options, onconnect).

Parameters

Pipe.createServer(opts?: PipeServerOptions, onconnection?: (pipe: Pipe) => void): PipeServer

Create a new pipe server. The server extends EventEmitter.

Overloads:

Pipe.createServer(opts?: PipeServerOptions, onconnection?: (pipe: Pipe) => void): PipeServer
Pipe.createServer(onconnection: (pipe: Pipe) => void): PipeServer

Parameters

Pipe.pipe(): [read: number, write: number]

Returns [read: number, write: number] — A [read, write] pair of file descriptors connected to each other.

push(data: unknown | null, encoding?: BufferEncoding): boolean

Parameters

read(): unknown | null

readable: boolean

readyState: 'open' | 'opening' | 'readOnly' | 'writeOnly' | 'closed'

The current state of the pipe. One of 'open', 'opening', 'readOnly', 'writeOnly', or 'closed'.

Pipe.ref(): this

Ref the pipe, preventing the process from exiting.

resume(): this

setEncoding(encoding: BufferEncoding): void

Parameters

uncork(): void

Pipe.unref(): this

Unref the pipe, allowing the process to exit.

unshift(data: unknown | null, encoding?: BufferEncoding): boolean

Parameters

writable: boolean

write

write(chunk: Buffer | string, encoding: BufferEncoding, handle?: IPCAcceptable, cb?: (err: Error | null) => void): boolean

Write chunk to the pipe. If handle is given and the pipe was created with ipc: true, the handle is transferred to the receiver alongside the chunk. handle must implement the IPCAcceptable protocol.

Parameters

PipeServer

new PipeServer(opts?: PipeServerOptions, onconnection?: (pipe: Pipe) => void)

Overloads:

new PipeServer(opts?: PipeServerOptions, onconnection?: (pipe: Pipe) => void)
new PipeServer(onconnection: (pipe: Pipe) => void)

Parameters

addListener<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

addOnceListener<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

address(): string | null

Returns string | null — The bound path, or null if the server is not listening.

close(onclose?: () => void): this

Close the server. No new connections will be accepted. The server emits close after all existing connections have ended.

Parameters

closing: boolean

emit<E extends keyof M>(name: E, ...args: M[E]): boolean

Parameters

eventNames(): (keyof M)[]

getMaxListeners(): number

listen

listen(path: string, backlog?: number, opts?: PipeServerListenOptions, onlistening?: () => void): this

Start listening for connections on path. backlog defaults to 511.

Parameters

Throws

• SERVER_ALREADY_LISTENING — the server is already listening.
• SERVER_IS_CLOSED — the server has been closed.

listenerCount<E extends keyof M>(name: E): number

Parameters

listeners<E extends keyof M>(name: E): EventHandler<M[E]>[]

Parameters

listening: boolean

Whether the server is listening.

off<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

on<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

once<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

prependListener<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

prependOnceListener<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

rawListeners<E extends keyof M>(name: E): EventHandler<M[E]>[]

Parameters

PipeServer.ref(): this

Ref the pipe, preventing the process from exiting.

removeAllListeners<E extends keyof M>(name?: E): this

Parameters

removeListener<E extends keyof M>(name: E, fn: EventHandler<M[E]>): this

Parameters

setMaxListeners(n: number): this

Parameters

PipeServer.unref(): this

Unref the pipe, allowing the process to exit.

Types

CreateConnectionOptions

interface CreateConnectionOptions {
  allowHalfOpen?: boolean
  eagerOpen?: boolean
  ipc?: boolean
  readBufferSize?: number
  path?: string
}

IPCAcceptable

interface IPCAcceptable {
  readonly [ipcHandle]: unknown
  [ipcAccept]?(): void
}

The protocol an object must implement to be passed to pipe.accept(target) or received via pipe.write(chunk, handle, ...); see IPCAcceptable protocol.

PipeConnectOptions

interface PipeConnectOptions {
  path?: string
}

PipeEvents

interface PipeEvents {
  connect: []
  handle: [type: number]
  data: [data: unknown]
  end: []
  readable: []
  piping: [dest: Writable]
  close: []
  error: [err: Error]
  drain: []
  finish: []
  pipe: [src: Readable]
}

PipeOpenOptions

interface PipeOpenOptions {
  fd?: number
}

PipeOptions

interface PipeOptions {
  allowHalfOpen?: boolean
  eagerOpen?: boolean
  ipc?: boolean
  readBufferSize?: number
}

PipeServerEvents

interface PipeServerEvents {
  close: []
  connection: [pipe: Pipe]
  error: [err: Error]
  listening: []
}

PipeServerListenOptions

interface PipeServerListenOptions {
  backlog?: number
  path?: string
}

PipeServerOptions

interface PipeServerOptions {
  allowHalfOpen?: boolean
  ipc?: boolean
  pauseOnConnect?: boolean
  readBufferSize?: number
}

Classes

PipeError

class PipeError {
  code: string
}

bare-pipe/constants

Constants and variables

constants

constants: {
  state: {
    CONNECTING: number
    CONNECTED: number
    BINDING: number
    BOUND: number
    READING: number
    CLOSING: number
    CLOSED: number
    UNREFED: number
    READABLE: number
    WRITABLE: number
  }
  path: {
    MAX_LENGTH: number
  }
  handle: {
    NAMED_PIPE: number
    TCP: number
    UDP: number
  }
}

Object containing internal state constants and handle type constants.

bare-pipe/errors

Classes

errors.PipeError

IPC handle passing

Pipes created with ipc: true can transfer libuv handles (named pipes, TCP sockets, UDP sockets) to a peer alongside the byte stream. The peer receives a 'handle' event for each transferred handle, in arrival order, before the corresponding 'data' event.

Handle passing needs a bidirectional socket on both ends, so the descriptors must come from a socket pair, such as bare-tcp's socketpair(). The descriptors from Pipe.pipe() are a unidirectional pipe and cannot carry handles.

Sender:

const left = new Pipe(fd, { ipc: true })
const socket = tcp.createConnection(port)

socket.on('connect', () => {
  left.write(Buffer.from('here'), socket)
})

Receiver:

const right = new Pipe(fd, { ipc: true })

right.on('handle', (type) => {
  if (type === Pipe.constants.handle.TCP) {
    const socket = new tcp.Socket()
    right.accept(socket)
    socket.on('data', console.log)
  }
})

IPCAcceptable protocol

Any object passed to pipe.accept(target) or pipe.write(chunk, handle, ...) must implement two well-known symbols:

const ipcHandle = Symbol.for('bare.ipc.handle')
const ipcAccept = Symbol.for('bare.ipc.accept')

class MyTarget {
  get [ipcHandle]() {
    return this._handle // An ArrayBuffer backing a libuv `uv_*_t` struct
  }

  [ipcAccept]() {
    // Optional: Called synchronously after the handle has been transferred
  }
}

• Symbol.for('bare.ipc.handle') (required): A getter returning the underlying libuv handle (typically an ArrayBuffer whose first bytes are a uv_stream_t / uv_udp_t).
• Symbol.for('bare.ipc.accept') (optional): A method invoked synchronously after the handle has been transferred. Use it to initialize per-handle state (for example address lookup).

Pipe, bare-tcp's Socket, bare-dgram's Socket, and any compatible package implement this protocol natively, so a bare-tcp socket can be passed and received via bare-pipe IPC without any glue code.

TypeScript users can import the IPCAcceptable interface from bare-pipe to type the protocol.

See also

• Builds on bare-events and bare-stream.
• bare-subprocess — uses pipes for child-process stdio.
• Bare modules — the full bare-* catalog.
• Bare runtime API — the runtime these modules extend.