From 0fcc248ac12eec9bf8cded3d30677787e7aeb28e Mon Sep 17 00:00:00 2001 From: Christopher Willis-Ford Date: Sat, 15 Jul 2017 23:51:59 -0700 Subject: [PATCH] Add tests for message dispatch system; fix bugs The tests run using TinyWorker, which emulates web workers on Node. There are quite a few quirks in that situation due to the differences between Node and Webpack as well as the differences between TinyWorker and real Web Workers. The tests also exposed a few bugs in the dispatch system, which have now been fixed. Most notably, if a method called through the dispatch system throws an exception that exception will now be passed back to the caller. Previously the exception would escape the dispatch system and the caller would never hear any response at all. --- package.json | 1 + src/dispatch/central-dispatch.js | 49 +++++++++-------- src/dispatch/worker-dispatch.js | 55 ++++++++----------- test/fixtures/dispatch-test-service.js | 20 +++++++ test/fixtures/dispatch-test-worker-shim.js | 19 +++++++ test/fixtures/dispatch-test-worker.js | 8 +++ test/unit/dispatch.js | 61 ++++++++++++++++++++++ 7 files changed, 160 insertions(+), 53 deletions(-) create mode 100644 test/fixtures/dispatch-test-service.js create mode 100644 test/fixtures/dispatch-test-worker-shim.js create mode 100644 test/fixtures/dispatch-test-worker.js create mode 100644 test/unit/dispatch.js diff --git a/package.json b/package.json index d20b8672..ec4cf86f 100644 --- a/package.json +++ b/package.json @@ -52,6 +52,7 @@ "socket.io-client": "1.7.3", "stats.js": "^0.17.0", "tap": "^10.2.0", + "tiny-worker": "^2.1.1", "webpack": "^2.4.1", "webpack-dev-server": "^2.4.1" } diff --git a/src/dispatch/central-dispatch.js b/src/dispatch/central-dispatch.js index 2eba7ecc..9d138872 100644 --- a/src/dispatch/central-dispatch.js +++ b/src/dispatch/central-dispatch.js @@ -12,7 +12,7 @@ class CentralDispatch { /** * List of callback registrations for promises waiting for a response from a call to a service on another * worker. A callback registration is an array of [resolve,reject] Promise functions. - * Calls to services on this worker don't enter this list. + * Calls to local services don't enter this list. * @type {Array.<[Function,Function]>} */ this.callbacks = []; @@ -32,6 +32,12 @@ class CentralDispatch { */ this.services = {}; + /** + * The constructor we will use to recognize workers. + * @type {Function} + */ + this.workerClass = (typeof Worker === 'undefined' ? null : Worker); + /** * List of workers attached to this dispatcher. * @type {Array} @@ -74,22 +80,26 @@ class CentralDispatch { */ transferCall (service, method, transfer, ...args) { return new Promise((resolve, reject) => { - if (this.services.hasOwnProperty(service)) { - const provider = this.services[service]; - if (provider instanceof Worker) { - const callbackId = this.nextCallback++; - this.callbacks[callbackId] = [resolve, reject]; - if (transfer) { - provider.postMessage([service, method, callbackId, args], transfer); + try { + if (this.services.hasOwnProperty(service)) { + const provider = this.services[service]; + if (provider instanceof this.workerClass) { + const callbackId = this.nextCallback++; + this.callbacks[callbackId] = [resolve, reject]; + if (transfer) { + provider.postMessage([service, method, callbackId, args], transfer); + } else { + provider.postMessage([service, method, callbackId, args]); + } } else { - provider.postMessage([service, method, callbackId, args]); + const result = provider[method].apply(provider, args); + resolve(result); } } else { - const result = provider[method].apply(provider, args); - resolve(result); + reject(new Error(`Service not found: ${service}`)); } - } else { - reject(new Error(`Service not found: ${service}`)); + } catch (e) { + reject(e); } }); } @@ -115,8 +125,8 @@ class CentralDispatch { addWorker (worker) { if (this.workers.indexOf(worker) === -1) { this.workers.push(worker); - worker.onmessage = this._onMessage.bind(this); - worker.postMessage('dispatch-handshake'); + worker.onmessage = this._onMessage.bind(this, worker); + worker.postMessage(['dispatch', '_handshake']); } else { log.warn('Ignoring attempt to add duplicate worker'); } @@ -124,11 +134,11 @@ class CentralDispatch { /** * Handle a message event received from a connected worker. + * @param {Worker} worker - the worker which sent the message. * @param {MessageEvent} event - the message event to be handled. * @private */ - _onMessage (event) { - const worker = event.target; + _onMessage (worker, event) { const [service, method, callbackId, ...args] = /** @type {[string, string, *]} */ event.data; if (service === 'dispatch') { switch (method) { @@ -167,7 +177,4 @@ class CentralDispatch { } } -const dispatch = new CentralDispatch(); -module.exports = dispatch; -self.Scratch = self.Scratch || {}; -self.Scratch.dispatch = dispatch; +module.exports = new CentralDispatch(); diff --git a/src/dispatch/worker-dispatch.js b/src/dispatch/worker-dispatch.js index e8563faf..a0127d06 100644 --- a/src/dispatch/worker-dispatch.js +++ b/src/dispatch/worker-dispatch.js @@ -12,7 +12,7 @@ class WorkerDispatch { /** * List of callback registrations for promises waiting for a response from a call to a service on another * worker. A callback registration is an array of [resolve,reject] Promise functions. - * Calls to services on this worker don't enter this list. + * Calls to local services don't enter this list. * @type {Array.<[Function,Function]>} */ this.callbacks = []; @@ -25,8 +25,6 @@ class WorkerDispatch { */ this._connectionPromise = new Promise(resolve => { this._onConnect = resolve; - }).then(() => { - self.onmessage = this._onMessage.bind(this); }); /** @@ -44,7 +42,10 @@ class WorkerDispatch { */ this.services = {}; - self.onmessage = this._onHandshake.bind(this); + this._onMessage = this._onMessage.bind(this); + if (typeof self !== 'undefined') { + self.onmessage = this._onMessage; + } } /** @@ -94,18 +95,22 @@ class WorkerDispatch { */ transferCall (service, method, transfer, ...args) { return new Promise((resolve, reject) => { - if (this.services.hasOwnProperty(service)) { - const provider = this.services[service]; - const result = provider[method].apply(provider, args); - resolve(result); - } else { - const callbackId = this.nextCallback++; - this.callbacks[callbackId] = [resolve, reject]; - if (transfer) { - self.postMessage([service, method, callbackId, ...args], transfer); + try { + if (this.services.hasOwnProperty(service)) { + const provider = this.services[service]; + const result = provider[method].apply(provider, args); + resolve(result); } else { - self.postMessage([service, method, callbackId, ...args]); + const callbackId = this.nextCallback++; + this.callbacks[callbackId] = [resolve, reject]; + if (transfer) { + self.postMessage([service, method, callbackId, ...args], transfer); + } else { + self.postMessage([service, method, callbackId, ...args]); + } } + } catch (e) { + reject(e); } }); } @@ -126,20 +131,6 @@ class WorkerDispatch { }); } - /** - * Message handler active until the dispatcher handshake arrives. - * @param {MessageEvent} event - the message event to be handled. - * @private - */ - _onHandshake (event) { - const message = event.data; - if (message === 'dispatch-handshake') { - this._onConnect(); - } else { - log.error(`WorkerDispatch received unexpected message before handshake: ${JSON.stringify(message)}`); - } - } - /** * Message handler active after the dispatcher handshake. This only handles method calls. * @param {MessageEvent} event - the message event to be handled. @@ -149,6 +140,9 @@ class WorkerDispatch { const [service, method, callbackId, ...args] = /** @type {[string, string, *]} */ event.data; if (service === 'dispatch') { switch (method) { + case '_handshake': + this._onConnect(); + break; case '_callback': this._callback(callbackId, ...args); break; @@ -184,7 +178,4 @@ class WorkerDispatch { } } -const dispatch = new WorkerDispatch(); -module.exports = dispatch; -self.Scratch = self.Scratch || {}; -self.Scratch.dispatch = dispatch; +module.exports = new WorkerDispatch(); diff --git a/test/fixtures/dispatch-test-service.js b/test/fixtures/dispatch-test-service.js new file mode 100644 index 00000000..1be99ea9 --- /dev/null +++ b/test/fixtures/dispatch-test-service.js @@ -0,0 +1,20 @@ +class DispatchTestService { + returnFortyTwo () { + return 42; + } + + doubleArgument (x) { + return 2 * x; + } + + throwException () { + throw new Error('This is a test exception thrown by LocalDispatchTest'); + } + + close () { + // eslint-disable-next-line no-undef + self.close(); + } +} + +module.exports = DispatchTestService; diff --git a/test/fixtures/dispatch-test-worker-shim.js b/test/fixtures/dispatch-test-worker-shim.js new file mode 100644 index 00000000..ae11a495 --- /dev/null +++ b/test/fixtures/dispatch-test-worker-shim.js @@ -0,0 +1,19 @@ +const Module = require('module'); + +const callsite = require('callsite'); +const path = require('path'); + +const oldRequire = Module.prototype.require; +Module.prototype.require = function (target) { + if (target.indexOf('/') === -1) { + return oldRequire.apply(this, arguments); + } + + const stack = callsite(); + const callerFile = stack[2].getFileName(); + const callerDir = path.dirname(callerFile); + target = path.resolve(callerDir, target); + return oldRequire.call(this, target); +}; + +oldRequire(path.resolve(__dirname, 'dispatch-test-worker')); diff --git a/test/fixtures/dispatch-test-worker.js b/test/fixtures/dispatch-test-worker.js new file mode 100644 index 00000000..4ca16a54 --- /dev/null +++ b/test/fixtures/dispatch-test-worker.js @@ -0,0 +1,8 @@ +const dispatch = require('../../src/dispatch/worker-dispatch'); +const DispatchTestService = require('./dispatch-test-service'); + +dispatch.setService('RemoteDispatchTest', new DispatchTestService()); + +dispatch.waitForConnection.then(() => { + dispatch.call('test', 'onWorkerReady'); +}); diff --git a/test/unit/dispatch.js b/test/unit/dispatch.js new file mode 100644 index 00000000..8e94735e --- /dev/null +++ b/test/unit/dispatch.js @@ -0,0 +1,61 @@ +const DispatchTestService = require('../fixtures/dispatch-test-service'); +const Worker = require('tiny-worker'); + +const dispatch = require('../../src/dispatch/central-dispatch'); +const path = require('path'); +const test = require('tap').test; + + +// By default Central Dispatch works with the Worker class built into the browser. Tell it to use TinyWorker instead. +dispatch.workerClass = Worker; + +const runServiceTest = function (serviceName, t) { + const promises = []; + + promises.push(dispatch.call(serviceName, 'returnFortyTwo').then(x => { + t.equal(x, 42); + })); + + promises.push(dispatch.call(serviceName, 'doubleArgument', 9).then(x => { + t.equal(x, 18); + })); + + promises.push(dispatch.call(serviceName, 'doubleArgument', 123).then(x => { + t.equal(x, 246); + })); + + // I tried using `t.rejects` here but ran into https://github.com/tapjs/node-tap/issues/384 + promises.push(dispatch.call(serviceName, 'throwException') + .then(() => { + t.fail('exception was not propagated as expected'); + }, () => { + t.pass('exception was propagated as expected'); + })); + + return Promise.all(promises); +}; + +test('local', t => { + dispatch.setService('LocalDispatchTest', new DispatchTestService()); + + return runServiceTest('LocalDispatchTest', t); +}); + +test('remote', t => { + const fixturesDir = path.resolve(__dirname, '../fixtures'); + const worker = new Worker('./test/fixtures/dispatch-test-worker-shim.js', null, {cwd: fixturesDir}); + dispatch.addWorker(worker); + + const waitForWorker = new Promise(resolve => { + dispatch.setService('test', { + onWorkerReady: resolve + }); + }); + + return waitForWorker + .then(() => runServiceTest('RemoteDispatchTest', t)) + .then(() => { + // Allow some time for the worker to finish, then terminate it + setTimeout(() => dispatch.call('RemoteDispatchTest', 'close'), 10); + }); +});