Protomux

Protomux multiplexes multiple message-oriented subprotocols over one framed stream. It is typically layered on top of Secretstream and uses Compact encoding for message schemas. For source and releases, see the Protomux repository.

Install

npm i protomux

Quickstart

import Protomux from 'protomux'
import cenc from 'compact-encoding'

const mux = new Protomux(aFramedEncryptedStream)

const channel = mux.createChannel({
  protocol: 'chat',
  onopen() {
    console.log('chat channel opened')
  }
})

const message = channel.addMessage({
  encoding: cenc.string,
  onmessage(text) {
    console.log(text)
  }
})

channel.open()
message.send('hello')

API Reference

Constructor and mux helpers

const mux = new Protomux(stream, [options])

• Signature: const mux = new Protomux(stream, [options])
• Parameters: stream is a framed stream that preserves message boundaries. options.alloc optionally overrides the buffer allocator used for outgoing frames.
• Returns: A Protomux instance bound to that stream.
• Example:

const mux = new Protomux(secretStream)

const mux = Protomux.from(streamOrMux, [options])

• Signature: const mux = Protomux.from(streamOrMux, [options])
• Parameters: streamOrMux is either an existing Protomux instance or a framed stream. options are forwarded when a new mux has to be created.
• Returns: An existing mux when one is already attached, otherwise a new Protomux.
• Example:

const mux = Protomux.from(secretStream)

const isMux = Protomux.isProtomux(value)

• Signature: const isMux = Protomux.isProtomux(value)
• Parameters: value is any unknown object.
• Returns: true when value looks like a Protomux instance.
• Example:

console.log(Protomux.isProtomux(mux))

mux.cork()

• Signature: mux.cork()
• Parameters: None.
• Returns: Nothing. Subsequent control and message frames are buffered into one batch until mux.uncork() runs.
• Example:

mux.cork()
messageA.send('a')
messageB.send('b')
mux.uncork()

mux.uncork()

• Signature: mux.uncork()
• Parameters: None.
• Returns: Nothing. Flushes the current batch when the cork depth returns to zero.
• Example:

mux.uncork()

const idle = mux.isIdle()

• Signature: const idle = mux.isIdle()
• Parameters: None.
• Returns: true when there are currently no open local channels.
• Example:

console.log(mux.isIdle())

for (const channel of mux)

• Signature: for (const channel of mux)
• Parameters: None.
• Returns: An iterator over the currently tracked local channels.
• Example:

for (const channel of mux) {
  console.log(channel.protocol)
}

Pairing and channel discovery

mux.pair({ protocol, [id] }, notify)

• Signature: mux.pair({ protocol, [id] }, notify)
• Parameters: protocol is the protocol name to match, id optionally narrows the match to one binary channel identifier, and notify is called whenever the remote side requests that channel.
• Returns: Nothing.
• Example:

mux.pair({ protocol: 'chat' }, async () => {
  console.log('remote requested chat')
})

mux.unpair({ protocol, [id] })

• Signature: mux.unpair({ protocol, [id] })
• Parameters: protocol and optional id identify the pairing callback to remove.
• Returns: Nothing.
• Example:

mux.unpair({ protocol: 'chat' })

const opened = mux.opened({ protocol, [id] })

• Signature: const opened = mux.opened({ protocol, [id] })
• Parameters: protocol and optional id identify the channel family to inspect.
• Returns: true if at least one matching channel is currently open.
• Example:

console.log(mux.opened({ protocol: 'chat' }))

const channel = mux.getLastChannel({ protocol, [id] })

• Signature: const channel = mux.getLastChannel({ protocol, [id] })
• Parameters: protocol and optional id identify the channel family to inspect.
• Returns: The most recently opened matching channel, or null if none is open.
• Example:

const latest = mux.getLastChannel({ protocol: 'chat' })

Creating channels

const channel = mux.createChannel(options)

• Signature: const channel = mux.createChannel(options)
• Parameters: options.protocol names the protocol, id optionally distinguishes multiple channels under one protocol, aliases adds alternate protocol names, unique defaults to true, handshake supplies an optional compact encoder for open-time handshake payloads, messages pre-registers message descriptors, userData stores arbitrary metadata, and onopen, onclose, ondestroy, and ondrain register lifecycle callbacks.
• Returns: A Channel instance, or null when a unique channel already exists or the remote has already rejected/closed the matching request.
• Example:

const channel = mux.createChannel({
  protocol: 'chat',
  handshake: cenc.string,
  onopen(handshake) {
    console.log(handshake)
  }
})

Mux properties

mux.stream

• Returns: The framed transport stream backing this mux.

mux.drained

• Returns: The current writable backpressure state from the underlying stream.

Channel lifecycle

channel.open([handshake])

• Signature: channel.open([handshake])
• Parameters: handshake is encoded with the channel's configured handshake encoder when one was supplied.
• Returns: Nothing. It allocates a local channel id and sends the open control message.
• Example:

channel.open('hello-handshake')

const opened = await channel.fullyOpened()

• Signature: const opened = await channel.fullyOpened()
• Parameters: None.
• Returns: A promise resolving to true when the channel fully opens or false if it closes first.
• Example:

channel.open()
const opened = await channel.fullyOpened()

await channel.fullyClosed()

• Signature: await channel.fullyClosed()
• Parameters: None.
• Returns: A promise resolving once the channel has been destroyed and all pending async handlers have finished.
• Example:

channel.close()
await channel.fullyClosed()

channel.close()

• Signature: channel.close()
• Parameters: None.
• Returns: Nothing. Sends a close control frame and marks the channel closed locally.
• Example:

channel.close()

channel.cork()

• Signature: channel.cork()
• Parameters: None.
• Returns: Nothing. Delegates to mux.cork() so this channel's writes can be batched.
• Example:

channel.cork()
message.send('a')
channel.uncork()

channel.uncork()

• Signature: channel.uncork()
• Parameters: None.
• Returns: Nothing. Delegates to mux.uncork().
• Example:

channel.uncork()

Registering and sending messages

const message = channel.addMessage([options])

• Signature: const message = channel.addMessage([options])
• Parameters: options.encoding is the compact encoder for the message payload, options.onmessage handles incoming payloads, and options.autoBatch defaults to true.
• Returns: A message descriptor with .send(...), .encoding, and .onmessage.
• Example:

const ping = channel.addMessage({
  encoding: cenc.string,
  onmessage(text) {
    console.log(text)
  }
})

const drained = message.send(data, [session])

• Signature: const drained = message.send(data, [session])
• Parameters: data is the payload to encode. session optionally targets another channel instance that uses the same message descriptor; it defaults to the channel the message was registered on.
• Returns: The mux backpressure boolean. It returns false if the target channel is already closed or if the underlying stream applied backpressure.
• Example:

ping.send('hello')

message.encoding

• Returns: The compact encoder associated with this message type.

message.onmessage

• Returns: The handler that runs when a message of this type arrives.

Channel properties

channel.userData

• Returns: The arbitrary metadata value passed to mux.createChannel(...).

channel.protocol

• Returns: The channel protocol name.

channel.aliases

• Returns: The list of alternate protocol names registered for this channel family.

channel.id

• Returns: The optional binary identifier distinguishing this channel family.

channel.handshake

• Returns: The decoded handshake payload after the channel fully opens, or null when no handshake encoder was configured.

channel.messages

• Returns: The array of registered message descriptors in type order.

channel.opened

• Returns: true after the channel has fully opened.

channel.closed

• Returns: true after either side closes the channel.

channel.destroyed

• Returns: true after the close lifecycle and pending handlers have fully drained.

channel.drained

• Returns: The current mux backpressure state for this channel's transport.

Lifecycle callbacks

onopen(handshake, channel)

• Signature: onopen(handshake, channel)
• Parameters: handshake is the decoded open payload or null. channel is the channel instance.
• Returns: The callback return value is ignored, but async promises are tracked and awaited before destruction.
• Example:

const channel = mux.createChannel({
  protocol: 'chat',
  onopen(handshake) {
    console.log('opened with', handshake)
  }
})

onclose(isRemote, channel)

• Signature: onclose(isRemote, channel)
• Parameters: isRemote is true when the remote side initiated the close. channel is the channel instance.
• Returns: The callback return value is ignored, but async promises are tracked.
• Example:

const channel = mux.createChannel({
  protocol: 'chat',
  onclose(isRemote) {
    console.log('closed by remote?', isRemote)
  }
})

ondestroy(channel)

• Signature: ondestroy(channel)
• Parameters: channel is the destroyed channel instance.
• Returns: The callback return value is ignored, but async promises are tracked.
• Example:

const channel = mux.createChannel({
  protocol: 'chat',
  ondestroy() {
    console.log('channel fully destroyed')
  }
})

ondrain(channel)

• Signature: ondrain(channel)
• Parameters: channel is the channel whose underlying mux became writable again.
• Returns: The callback return value is ignored, but async promises are tracked.
• Example:

const channel = mux.createChannel({
  protocol: 'chat',
  ondrain() {
    console.log('ready for more writes')
  }
})

See also

• Secretstream — the encrypted framed stream most commonly used underneath Protomux.
• Compact encoding — the schema toolkit Protomux uses for handshakes and message payloads.
• Hyperswarm — common peer-transport entry point before you layer on Secretstream and Protomux.
• Upstream Protomux repository — source, releases, and implementation details.