5445117558

const duplexify = require('duplexify')
const lpstream = require('length-prefixed-stream')
const through = require('through2')
const pump = require('pump')
const msgpack = require('@msgpack/msgpack')

/** @typedef {{encode: (object: any) => Buffer, decode: (data: Buffer) => any}} Encoding */

const extensionCodec = new msgpack.ExtensionCodec()

// msgpack defaults to decoding buffers as Uint8Array. Add extension to decode
// them to buffers
const BUF_EXT_TYPE = 0
extensionCodec.register({
  type: BUF_EXT_TYPE,
  encode: (object) => {
    if (Buffer.isBuffer(object)) {
      return msgpack.encode(object)
    } else {
      return null
    }
  },
  decode: (data) => {
    const uint8Array = /** @type {Uint8Array} */ (msgpack.decode(data))
    return Buffer.from(uint8Array)
  },
})

/** @type {Encoding} */
const defaultEncoding = {
  encode: (object) => Buffer.from(msgpack.encode(object, { extensionCodec })),
  decode: (data) => msgpack.decode(data, { extensionCodec }),
}

/**
 * @param {import('stream').Duplex} duplex Duplex stream, e.g. a socket/http connection
 * @param {object} [options]
 * @param {Encoding} [options.encoding=defaultEncoding] An encoding object that contains
 * encode(value) and decode(buffer) functions for encoding values to and from
 * buffers. Defaults to msgpack encoding with support for encoding/decoding
 * Buffers
 * @returns {import('stream').Duplex} Duplex stream in objectMode which can
 * write/read from and rpc client / server
 */
function createEncodeDecodeStream(duplex, options = {}) {
  const { encode, decode } = options.encoding || defaultEncoding

  const encodeStream = through.obj((chunk, enc, cb) => cb(null, encode(chunk)))
  const decodeStream = through.obj((chunk, enc, cb) => cb(null, decode(chunk)))

  const unEncodedStream = duplexify(encodeStream, decodeStream, {
    objectMode: true,
  })

  pump(
    encodeStream,
    lpstream.encode(),
    duplex,
    lpstream.decode(),
    decodeStream,
    (err) => {
      unEncodedStream.destroy(err)
    }
  )

  return unEncodedStream
}

module.exports = createEncodeDecodeStream

1	const duplexify = require('duplexify')	12✔
2	const lpstream = require('length-prefixed-stream')	12✔
3	const through = require('through2')	12✔
4	const pump = require('pump')	12✔
5	const msgpack = require('@msgpack/msgpack')	12✔
6
7	/** @typedef {{encode: (object: any) => Buffer, decode: (data: Buffer) => any}} Encoding */
8
9	const extensionCodec = new msgpack.ExtensionCodec()	12✔
10
11	// msgpack defaults to decoding buffers as Uint8Array. Add extension to decode
12	// them to buffers
13	const BUF_EXT_TYPE = 0	12✔
14	extensionCodec.register({	12✔
15	type: BUF_EXT_TYPE,
16	encode: (object) => {
17	if (Buffer.isBuffer(object)) {	102✔
18	return msgpack.encode(object)	6✔
19	} else {
20	return null	96✔
21	}
22	},
23	decode: (data) => {
24	const uint8Array = /** @type {Uint8Array} */ (msgpack.decode(data))	6✔
25	return Buffer.from(uint8Array)	6✔
26	},
27	})
28
29	/** @type {Encoding} */
30	const defaultEncoding = {	12✔
31	encode: (object) => Buffer.from(msgpack.encode(object, { extensionCodec })),	42✔
32	decode: (data) => msgpack.decode(data, { extensionCodec }),	42✔
33	}
34
35	/**
36	* @param {import('stream').Duplex} duplex Duplex stream, e.g. a socket/http connection
37	* @param {object} [options]
38	* @param {Encoding} [options.encoding=defaultEncoding] An encoding object that contains
39	* encode(value) and decode(buffer) functions for encoding values to and from
40	* buffers. Defaults to msgpack encoding with support for encoding/decoding
41	* Buffers
42	* @returns {import('stream').Duplex} Duplex stream in objectMode which can
43	* write/read from and rpc client / server
44	*/
45	function createEncodeDecodeStream(duplex, options = {}) {	3✔
46	const { encode, decode } = options.encoding \|\| defaultEncoding	6✔
47
48	const encodeStream = through.obj((chunk, enc, cb) => cb(null, encode(chunk)))	45✔
49	const decodeStream = through.obj((chunk, enc, cb) => cb(null, decode(chunk)))	45✔
50
51	const unEncodedStream = duplexify(encodeStream, decodeStream, {	6✔
52	objectMode: true,
53	})
54
55	pump(	6✔
56	encodeStream,
57	lpstream.encode(),
58	duplex,
59	lpstream.decode(),
60	decodeStream,
61	(err) => {
62	unEncodedStream.destroy(err)	6✔
63	}
64	)
65
66	return unEncodedStream	6✔
67	}
68
69	module.exports = createEncodeDecodeStream	12✔

gmaclennan / rpc-reflector / 5445117558

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous