/** * Copyright 2017 Google Inc. All rights reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ const debugProtocol = require('debug')('hdfox:protocol'); const EventEmitter = require('events'); /** * @internal */ class Connection extends EventEmitter { /** * @param {!Puppeteer.ConnectionTransport} transport * @param {number=} delay */ constructor(transport, delay = 0) { super(); this._lastId = 0; /** @type {!Map}*/ this._callbacks = new Map(); this._delay = delay; this._transport = transport; this._transport.onmessage = this._onMessage.bind(this); this._transport.onclose = this._onClose.bind(this); this._closed = false; } /** * @param {string} method * @param {!Object=} params * @return {!Promise} */ send(method, params = {}) { const id = ++this._lastId; const message = JSON.stringify({id, method, params}); debugProtocol('SEND ► ' + message); this._transport.send(message); return new Promise((resolve, reject) => { this._callbacks.set(id, {resolve, reject, error: new Error(), method}); }); } /** * @param {string} message */ async _onMessage(message) { if (this._delay) await new Promise(f => setTimeout(f, this._delay)); debugProtocol('◀ RECV ' + message); const object = JSON.parse(message); if (object.id) { const callback = this._callbacks.get(object.id); // Callbacks could be all rejected if someone has called `.dispose()`. if (callback) { this._callbacks.delete(object.id); if (object.error) callback.reject(createProtocolError(callback.error, callback.method, object)); else callback.resolve(object.result); } } else { this.emit(object.method, object.params); } } _onClose() { if (this._closed) return; this._closed = true; this._transport.onmessage = null; this._transport.onclose = null; for (const callback of this._callbacks.values()) callback.reject(rewriteError(callback.error, `Protocol error (${callback.method}): Target closed.`)); this._callbacks.clear(); this.emit(Connection.Events.Disconnected); } dispose() { this._onClose(); this._transport.close(); } } Connection.Events = { Disconnected: Symbol('Connection.Events.Disconnected'), }; /** * @param {!Error} error * @param {string} method * @param {{error: {message: string, data: any}}} object * @return {!Error} */ function createProtocolError(error, method, object) { let message = `Protocol error (${method}): ${object.error.message}`; if ('data' in object.error) message += ` ${object.error.data}`; return rewriteError(error, message); } /** * @param {!Error} error * @param {string} message * @return {!Error} */ function rewriteError(error, message) { error.message = message; return error; } module.exports = {Connection};