mplex2

A stream multiplexer written in ES6.

MPlex is a duplex stream with the following additional events:

• ready: Emitted when the stream fully initialized itself. Call API methods only after this event.
• stream: The remote side requested a stream.
  • options: Data sent by the remote when requesting this stream
  • fn(reason): A function to be called with a string as a reason of declining the stream request, or without parameters. In the letter case the request is accepted, and the function returns an MStream instance ready to read and write
• globalRequest: The remote side issued a global request.
  • data: The request as set by the remote
  • fn(success, details): A function to be called to send the response. success can be true or false, details is some JSON serializable data.
• close: No more action on the stream. Will be emitted after an error.

API:

• new MPlex([options])
  • options <Object>
    • highWaterMark: The same as with streams (16384)
    • wantMaxPacketSize: The maximum size of a packet the remote side is allowed to send (8192). The minimum is 1024. Handled automatically
    • wantContinueAfter: After sending this amount of data to an underlying stream, MPlex will not send more data to this stream unless an explicit continue instruction on the stream. Handled automatically, see continueLevel
    • continueLevel: The stream will send continue to the remote only when data in its' buffer is less than this amount
    • streamHighWaterMark: highWaterMark for underlying streams
    • maxID: The maximum number of open streams in MPlex
• MPlex.newStream(options, callback):
  • options: Any JSON serializable data, will be sent to the remote
  • callback(error, stream): Called when the request is accepted or rejected by the remote.
    • error: an error instance indicating the reason of decline
    • stream: an MStream instance (see below) which is ready for reding and writing
• MPlex.globalRequest(data, callback): x
  • data: JSON serializable data to send
  • callback(error, sucess, details): will be called when a response arrives (or if error occures during sending).
    • error: If not null, an error occured during sending the request
    • success: boolean value received from the remote, indicating the status of the request.
    • details: additional data from the remote.
• MPlex.close(): Close MPlex.

MStream is a duplex stream returned by MPlex.newStream or the second parameter (a function, fn) of the stream event. Apart from the standard stream events, it also emits close, which indicates no more processing will happen on this stream.

API:

• MStream.close(): Close this stream.

HeartBeatMPlex is an extension to MPlex. It accepts the following extra options:

• wantHeartBeatMilli: Requests the remote side not to be idle for longer than this amount in milliseconds.
• heartBeatThreshold: If the remote is idle for more than wantHeartBeatMilli + heartBeatThreshold milliseconds, an error will be emitted.

A basic example

const {MPlex} = require('mplex2')

const mplex1 = new MPlex()
const mplex2 = new MPlex()

mplex1.on('ready', () => {
  mplex1.newStream('simple stream', (err, stream) => {
    if (err) console.log('rejected')
    if (stream) {
      stream.end('message to remote')
      stream.once('finish', stream.close)
    }
  })
})

mplex2.on('stream', (options, fn) => {
  const stream = fn()  // accept
  stream.on('data', d => console.log(d.toString()))
  stream.on('close', () => console.log('remote closed'))
})

mplex1.pipe(mplex2).pipe(mplex1)