5280286787

pending completion

Build # 5280286787

Build Type

push

github

Committed by

gmaclennan

Commit Message

chore(release): 1.3.4

Run Details

190 of 203 branches covered (93.6%)

Branch coverage included in aggregate %.

320 of 328 relevant lines covered (97.56%)

749.59 hits per line

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

94.27

/server.js

const assert = require('assert')
const { serializeError } = require('serialize-error')

const isStream = require('is-stream')
const { msgType } = require('./lib/constants')
const isValidMessage = require('./lib/validate-message')
const { parse, stringify } = require('./lib/prop-array-utils')
const MessageStream = require('./lib/message-stream')
const isMessagePortLike = require('./lib/is-message-port-like')
const { EventEmitter } = require('events')

/** @typedef {import('./lib/types').MsgRequest} MsgRequest */
/** @typedef {import('./lib/types').MsgResponse} MsgResponse */
/** @typedef {import('./lib/types').MsgOn} MsgOn */
/** @typedef {import('./lib/types').MsgOff} MsgOff */
/** @typedef {import('./lib/types').MsgEmit} MsgEmit */
/** @typedef {import('./lib/types').Message} Message */
/** @typedef {import('./lib/types').NonEmptyArray<string>} NonEmptyStringArray */
/** @typedef {import('./lib/types').MessagePortLike} MessagePortLike */
/** @typedef {import('worker_threads').MessagePort} MessagePortNode */

module.exports = createServer

/**
 * @public
 * Create an RPC server that will receive messages via `receiver`, call the
 * matching method on `handler`, and send the reply via `send`.
 *
 * @param {{[method: string]: any}} handler Any method called on the client
 * object will be called on this object. Methods can return a value, a Promise,
 * or a ReadableStream. Your transport stream must be able to encode/decode any
 * values that your handler returns
 * @param {MessagePort | MessagePortLike | MessagePortNode | import('stream').Duplex} channel A Duplex Stream with objectMode=true or a MessagePort-like object that must implement an `.on('message')` event handler and a `.postMessage()` method.
 *
 * @returns {{ close: () => void }} An object with a single method `close()`
 * that will stop the server listening to and sending any more messages
 */
function createServer(handler, channel) {
  assert(typeof handler === 'object', 'Missing handler object.')
  const channelIsStream = isStream.duplex(channel)
  assert(
    isMessagePortLike(channel) || channelIsStream,
    'Must pass a Duplex Stream or a browser MessagePort, node worker.MessagePort, or MessagePort-like object'
  )

  /** @type {Map<string, (...args: any[]) => void>} */
  let subscriptions = new Map()

  if (channelIsStream) {
    channel.on('data', handleMessage)
  } else if ('on' in channel) {
    channel.on('message', handleMessage)
  } else {
    channel.addEventListener('message', handleMessage)
  }

  /** @param {MsgResponse | MsgEmit} msg */
  function send(msg) {
    // TODO: Do we need back pressure here? Would just result in buffering here
    // vs. buffering in the stream, so probably no
    if (channelIsStream) {
      channel.write(msg)
    } else {
      channel.postMessage(msg)
    }
  }

  /**
   * Handles an incoming message.
   * @param {unknown} msg Can be any type, but we only process messages types that
   * we understand, other messages are ignored
   */
  function handleMessage(msg) {
    // When using a MessagePort in a browser or electron environment, the
    // actual data is in `event.data`
    if (typeof msg === 'object' && msg && 'data' in msg) {
      msg = msg.data
    }
    if (!isValidMessage(msg)) return
    switch (msg[0]) {
      case msgType.REQUEST:
        return handleRequest(/** @type {MsgRequest} */ (msg))
      case msgType.ON:
        return handleOn(/** @type {MsgOn} */ (msg))
      case msgType.OFF:
        return handleOff(/** @type {MsgOff} */ (msg))
      default:
        console.warn(`Unhandled message type: ${msg[0]}. (Message was ignored)`)
    }
  }

  /** @param {MsgRequest} msg */
  async function handleRequest([, msgId, propArray, args]) {
    /** @type {MsgResponse} */
    let response

    try {
      const result = await Promise.resolve(
        applyNestedMethod(handler, propArray, args)
      )
      if (isStream.readable(result)) {
        return handleStream(result)
      }
      response = [msgType.RESPONSE, msgId, null, result]
    } catch (error) {
      response = [msgType.RESPONSE, msgId, serializeError(error)]
    }

    send(response)

    /** @param {import('stream').Readable} stream */
    function handleStream(stream) {
      const rs = stream.pipe(new MessageStream(msgId))
      if (channelIsStream) {
        rs.pipe(channel, { end: false })
      } else {
        rs.on('data', (chunk) => send(chunk))
      }
      rs.on('error', (err) =>
        send([msgType.RESPONSE, msgId, serializeError(err)])
      )
    }
  }

  /** @param {MsgOn} msg */
  function handleOn([, eventName, propArray]) {
    let emitter
    try {
      emitter = getNestedEventEmitter(handler, propArray)
    } catch (e) {
      return console.warn(e + '. (Subscription from client was ignored)')
    }

    const encodedEventName = stringify(propArray, eventName)

    // If we are already emitting for this event, we can ignore
    if (subscriptions.has(encodedEventName)) return

    /** @type {(...args: any[]) => void} */
    const listener = (...args) => {
      if (args.length === 1 && args[0] instanceof Error) {
        send([msgType.EMIT, eventName, propArray, serializeError(args[0])])
      } else {
        send([msgType.EMIT, eventName, propArray, null, args])
      }
    }
    subscriptions.set(encodedEventName, listener)
    emitter.on(eventName, listener)
  }

  /** @param {MsgOff} msg */
  function handleOff([, eventName, propArray]) {
    let emitter
    try {
      emitter = getNestedEventEmitter(handler, propArray)
    } catch (e) {
      return console.warn(e + '. (Subscription from client was ignored)')
    }

    const encodedEventName = stringify(propArray, eventName)

    // Fail silently if there is nothing to unsubscribe
    if (!subscriptions.has(encodedEventName)) return

    const listener = subscriptions.get(encodedEventName)
    listener && emitter.removeListener(eventName, listener)
    subscriptions.delete(encodedEventName)
  }

  return {
    close: () => {
      if (channelIsStream) {
        channel.off('data', handleMessage)
      } else if ('off' in channel) {
        channel.off('message', handleMessage)
      } else {
        channel.removeEventListener('message', handleMessage)
      }
      for (const [encodedEventName, listener] of subscriptions.entries()) {
        const [propArray, eventName] = parse(encodedEventName)
        try {
          const emitter = getNestedEventEmitter(handler, propArray)
          emitter.removeListener(eventName, listener)
        } catch (e) {}
      }
      subscriptions = new Map()
    },
  }
}

/**
 * @private
 * Calls a deeply nested property function. Throws a TypeError if not a function
 *
 * @param {{[propertyKey: string]: any}} target
 * @param {NonEmptyStringArray} propArray
 * @param {ArrayLike<any>} args
 * @returns {any}
 */
function applyNestedMethod(target, propArray, args) {
  let nested = target
  for (const propertyKey of propArray.slice(0, -1)) {
    if (!Reflect.has(nested, propertyKey)) {
      throw new ReferenceError(`${propertyKey} is not defined`)
    }
    nested = nested[propertyKey]
  }
  const propertyKey = propArray[propArray.length - 1]
  if (nested === null) {
    throw new TypeError(`Cannot read property '${propertyKey}' of null`)
  }
  if (typeof nested === 'object') {
    if (!Reflect.has(nested, propertyKey)) {
      throw new ReferenceError(`${propertyKey} is not defined`)
    }
  } else if (typeof nested[propertyKey] === 'undefined') {
    throw new ReferenceError(`${propertyKey} is not defined`)
  }
  if (typeof nested[propertyKey] === 'function') {
    return Reflect.apply(nested[propertyKey], nested, args)
  }
  if (typeof nested[propertyKey] === 'symbol') {
    throw new TypeError(`Property '${propertyKey}' is a Symbol`)
  }
  return nested[propertyKey]
}

/**
 * @private
 * Returns a deeply nested event emitter
 *
 * @param {{[propertyKey: string]: any}} target
 * @param {string[]} propArray
 * @returns {EventEmitter}
 */
function getNestedEventEmitter(target, propArray) {
  let nested = target
  for (const propertyKey of propArray) {
    if (!Reflect.has(nested, propertyKey)) {
      throw new ReferenceError(`${propertyKey} is not defined`)
    }
    nested = nested[propertyKey]
  }
  if (!(nested instanceof EventEmitter)) {
    throw new TypeError(
      `${
        propArray.length === 0 ? '[target]' : propArray[propArray.length - 1]
      } is not an EventEmitter`
    )
  }
  return nested
}

1	const assert = require('assert')	9✔
2	const { serializeError } = require('serialize-error')	9✔
3
4	const isStream = require('is-stream')	9✔
5	const { msgType } = require('./lib/constants')	9✔
6	const isValidMessage = require('./lib/validate-message')	9✔
7	const { parse, stringify } = require('./lib/prop-array-utils')	9✔
8	const MessageStream = require('./lib/message-stream')	9✔
9	const isMessagePortLike = require('./lib/is-message-port-like')	9✔
10	const { EventEmitter } = require('events')	9✔
11
12	/** @typedef {import('./lib/types').MsgRequest} MsgRequest */
13	/** @typedef {import('./lib/types').MsgResponse} MsgResponse */
14	/** @typedef {import('./lib/types').MsgOn} MsgOn */
15	/** @typedef {import('./lib/types').MsgOff} MsgOff */
16	/** @typedef {import('./lib/types').MsgEmit} MsgEmit */
17	/** @typedef {import('./lib/types').Message} Message */
18	/** @typedef {import('./lib/types').NonEmptyArray<string>} NonEmptyStringArray */
19	/** @typedef {import('./lib/types').MessagePortLike} MessagePortLike */
20	/** @typedef {import('worker_threads').MessagePort} MessagePortNode */
21
22	module.exports = createServer	9✔
23
24	/**
25	* @public
26	* Create an RPC server that will receive messages via `receiver`, call the
27	* matching method on `handler`, and send the reply via `send`.
28	*
29	* @param {{[method: string]: any}} handler Any method called on the client
30	* object will be called on this object. Methods can return a value, a Promise,
31	* or a ReadableStream. Your transport stream must be able to encode/decode any
32	* values that your handler returns
33	* @param {MessagePort \| MessagePortLike \| MessagePortNode \| import('stream').Duplex} channel A Duplex Stream with objectMode=true or a MessagePort-like object that must implement an `.on('message')` event handler and a `.postMessage()` method.
34	*
35	* @returns {{ close: () => void }} An object with a single method `close()`
36	* that will stop the server listening to and sending any more messages
37	*/
38	function createServer(handler, channel) {
39	assert(typeof handler === 'object', 'Missing handler object.')	135✔
40	const channelIsStream = isStream.duplex(channel)	135✔
41	assert(	135✔
42	isMessagePortLike(channel) \|\| channelIsStream,	204✔
43	'Must pass a Duplex Stream or a browser MessagePort, node worker.MessagePort, or MessagePort-like object'
44	)
45
46	/** @type {Map<string, (...args: any[]) => void>} */
47	let subscriptions = new Map()	135✔
48
49	if (channelIsStream) {	135✔
50	channel.on('data', handleMessage)	69✔
51	} else if ('on' in channel) {	66!
52	channel.on('message', handleMessage)	66✔
53	} else {
54	channel.addEventListener('message', handleMessage)	×
55	}
56
57	/** @param {MsgResponse \| MsgEmit} msg */
58	function send(msg) {
59	// TODO: Do we need back pressure here? Would just result in buffering here
60	// vs. buffering in the stream, so probably no
61	if (channelIsStream) {	4,380✔
62	channel.write(msg)	108✔
63	} else {
64	channel.postMessage(msg)	4,272✔
65	}
66	}
67
68	/**
69	* Handles an incoming message.
70	* @param {unknown} msg Can be any type, but we only process messages types that
71	* we understand, other messages are ignored
72	*/
73	function handleMessage(msg) {
74	// When using a MessagePort in a browser or electron environment, the
75	// actual data is in `event.data`
76	if (typeof msg === 'object' && msg && 'data' in msg) {	441!
77	msg = msg.data	×
78	}
79	if (!isValidMessage(msg)) return	441✔
80	switch (msg[0]) {	348✔
81	case msgType.REQUEST:
82	return handleRequest(/** @type {MsgRequest} */ (msg))	210✔
83	case msgType.ON:
84	return handleOn(/** @type {MsgOn} */ (msg))	84✔
85	case msgType.OFF:
86	return handleOff(/** @type {MsgOff} */ (msg))	30✔
87	default:
88	console.warn(`Unhandled message type: ${msg[0]}. (Message was ignored)`)	24✔
89	}
90	}
91
92	/** @param {MsgRequest} msg */
93	async function handleRequest([, msgId, propArray, args]) {
94	/** @type {MsgResponse} */
95	let response
96
97	try {	210✔
98	const result = await Promise.resolve(	210✔
99	applyNestedMethod(handler, propArray, args)
100	)
101	if (isStream.readable(result)) {	126✔
102	return handleStream(result)	36✔
103	}
104	response = [msgType.RESPONSE, msgId, null, result]	90✔
105	} catch (error) {
106	response = [msgType.RESPONSE, msgId, serializeError(error)]	84✔
107	}
108
109	send(response)	174✔
110
111	/** @param {import('stream').Readable} stream */
112	function handleStream(stream) {
113	const rs = stream.pipe(new MessageStream(msgId))	36✔
114	if (channelIsStream) {	36✔
115	rs.pipe(channel, { end: false })	18✔
116	} else {
117	rs.on('data', (chunk) => send(chunk))	4,164✔
118	}
119	rs.on('error', (err) =>	36✔
120	send([msgType.RESPONSE, msgId, serializeError(err)])	×
121	)
122	}
123	}
124
125	/** @param {MsgOn} msg */
126	function handleOn([, eventName, propArray]) {
127	let emitter
128	try {	84✔
129	emitter = getNestedEventEmitter(handler, propArray)	84✔
130	} catch (e) {
131	return console.warn(e + '. (Subscription from client was ignored)')	12✔
132	}
133
134	const encodedEventName = stringify(propArray, eventName)	72✔
135
136	// If we are already emitting for this event, we can ignore
137	if (subscriptions.has(encodedEventName)) return	72✔
138
139	/** @type {(...args: any[]) => void} */
140	const listener = (...args) => {	66✔
141	if (args.length === 1 && args[0] instanceof Error) {	42✔
142	send([msgType.EMIT, eventName, propArray, serializeError(args[0])])	6✔
143	} else {
144	send([msgType.EMIT, eventName, propArray, null, args])	36✔
145	}
146	}
147	subscriptions.set(encodedEventName, listener)	66✔
148	emitter.on(eventName, listener)	66✔
149	}
150
151	/** @param {MsgOff} msg */
152	function handleOff([, eventName, propArray]) {
153	let emitter
154	try {	30✔
155	emitter = getNestedEventEmitter(handler, propArray)	30✔
156	} catch (e) {
157	return console.warn(e + '. (Subscription from client was ignored)')	12✔
158	}
159
160	const encodedEventName = stringify(propArray, eventName)	18✔
161
162	// Fail silently if there is nothing to unsubscribe
163	if (!subscriptions.has(encodedEventName)) return	18✔
164
165	const listener = subscriptions.get(encodedEventName)	12✔
166	listener && emitter.removeListener(eventName, listener)	12✔
167	subscriptions.delete(encodedEventName)	12✔
168	}
169
170	return {	135✔
171	close: () => {
172	if (channelIsStream) {	18✔
173	channel.off('data', handleMessage)	9✔
174	} else if ('off' in channel) {	9!
175	channel.off('message', handleMessage)	9✔
176	} else {
177	channel.removeEventListener('message', handleMessage)	×
178	}
179	for (const [encodedEventName, listener] of subscriptions.entries()) {	18✔
180	const [propArray, eventName] = parse(encodedEventName)	24✔
181	try {	24✔
182	const emitter = getNestedEventEmitter(handler, propArray)	24✔
183	emitter.removeListener(eventName, listener)	24✔
184	} catch (e) {}
185	}
186	subscriptions = new Map()	18✔
187	},
188	}
189	}
190
191	/**
192	* @private
193	* Calls a deeply nested property function. Throws a TypeError if not a function
194	*
195	* @param {{[propertyKey: string]: any}} target
196	* @param {NonEmptyStringArray} propArray
197	* @param {ArrayLike<any>} args
198	* @returns {any}
199	*/
200	function applyNestedMethod(target, propArray, args) {
201	let nested = target	210✔
202	for (const propertyKey of propArray.slice(0, -1)) {	210✔
203	if (!Reflect.has(nested, propertyKey)) {	96✔
204	throw new ReferenceError(`${propertyKey} is not defined`)	6✔
205	}
206	nested = nested[propertyKey]	90✔
207	}
208	const propertyKey = propArray[propArray.length - 1]	204✔
209	if (nested === null) {	204✔
210	throw new TypeError(`Cannot read property '${propertyKey}' of null`)	6✔
211	}
212	if (typeof nested === 'object') {	198✔
213	if (!Reflect.has(nested, propertyKey)) {	168✔
214	throw new ReferenceError(`${propertyKey} is not defined`)	30✔
215	}
216	} else if (typeof nested[propertyKey] === 'undefined') {	30!
217	throw new ReferenceError(`${propertyKey} is not defined`)	24✔
218	}
219	if (typeof nested[propertyKey] === 'function') {	138✔
220	return Reflect.apply(nested[propertyKey], nested, args)	84✔
221	}
222	if (typeof nested[propertyKey] === 'symbol') {	54✔
223	throw new TypeError(`Property '${propertyKey}' is a Symbol`)	6✔
224	}
225	return nested[propertyKey]	48✔
226	}
227
228	/**
229	* @private
230	* Returns a deeply nested event emitter
231	*
232	* @param {{[propertyKey: string]: any}} target
233	* @param {string[]} propArray
234	* @returns {EventEmitter}
235	*/
236	function getNestedEventEmitter(target, propArray) {
237	let nested = target	138✔
238	for (const propertyKey of propArray) {	138✔
239	if (!Reflect.has(nested, propertyKey)) {	42✔
240	throw new ReferenceError(`${propertyKey} is not defined`)	12✔
241	}
242	nested = nested[propertyKey]	30✔
243	}
244	if (!(nested instanceof EventEmitter)) {	126✔
245	throw new TypeError(	12✔
246	`${
247	propArray.length === 0 ? '[target]' : propArray[propArray.length - 1]	12!
248	} is not an EventEmitter`
249	)
250	}
251	return nested	114✔
252	}

gmaclennan / rpc-reflector / 5280286787

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