feat: Integrate stdio transport from upstream PR #440 (--remote-debugging-pipe)

README.md

@@ -456,8 +456,12 @@ Connects to a remote instance using the [Chrome Debugging Protocol].
 - `protocol`: [Chrome Debugging Protocol] descriptor object. Defaults to use the
   protocol chosen according to the `local` option;
 - `local`: a boolean indicating whether the protocol must be fetched *remotely*
-  or if the local version must be used. It has no effect if the `protocol`
-  option is set. Defaults to `false`.
+  or if the local version must be used. It has no effect if the `protocol` or
+  `process` option is set. Defaults to `false`.
+- `process`: a `ChildProcess` object that represents a Chrome instance launched
+  with `--remote-debugging-pipe`. If passed, websocket-related options will be
+  ignored and communications will occur over stdio instead. Note: the `protocol`
+  cannot be fetched remotely if a `process` is passed.
 
 These options are also valid properties of all the instances of the `CDP`
 class. In addition to that, the `webSocketUrl` field contains the currently used

lib/chrome.js

@@ -10,6 +10,7 @@ const WebSocket = require('ws');
 const api = require('./api.js');
 const defaults = require('./defaults.js');
 const devtools = require('./devtools.js');
+const StdioWrapper = require('./stdio-wrapper.js');
 
 class ProtocolError extends Error {
     constructor(request, response) {
@@ -58,6 +59,7 @@ class Chrome extends EventEmitter {
         this.local = !!(options.local);
         this.target = options.target || defaultTarget;
         this.connectOptions = options.connectOptions;
+        this.process = options.process;
         // locals
         this._notifier = notifier;
         this._callbacks = {};
@@ -92,7 +94,7 @@ class Chrome extends EventEmitter {
                         const request = {method, params, sessionId};
                         reject(
                             error instanceof Error
-                                ? error // low-level WebSocket error
+                                ? error // low-level io error
                                 : new ProtocolError(request, response)
                         );
                     } else {
@@ -104,56 +106,52 @@ class Chrome extends EventEmitter {
     }
 
     close(callback) {
-        const closeWebSocket = (callback) => {
-            // don't close if it's already closed
-            if (this._ws.readyState === 3) {
-                callback();
-            } else {
-                // don't notify on user-initiated shutdown ('disconnect' event)
-                this._ws.removeAllListeners('close');
-                this._ws.once('close', () => {
-                    this._ws.removeAllListeners();
-                    this._handleConnectionClose();
-                    callback();
-                });
-                this._ws.close();
-            }
-        };
         if (typeof callback === 'function') {
-            closeWebSocket(callback);
+            this._close(callback);
             return undefined;
         } else {
             return new Promise((fulfill, reject) => {
-                closeWebSocket(fulfill);
+                this._close(fulfill);
             });
         }
     }
 
     // initiate the connection process
     async _start() {
-        const options = {
-            host: this.host,
-            port: this.port,
-            secure: this.secure,
-            useHostName: this.useHostName,
-            alterPath: this.alterPath,
-            ...this.connectOptions,
-        };
         try {
-            // fetch the WebSocket debugger URL
-            const url = await this._fetchDebuggerURL(options);
-            // allow the user to alter the URL
-            const urlObject = parseUrl(url);
-            urlObject.pathname = options.alterPath(urlObject.pathname);
-            this.webSocketUrl = formatUrl(urlObject);
-            // update the connection parameters using the debugging URL
-            options.host = urlObject.hostname;
-            options.port = urlObject.port || options.port;
-            // fetch the protocol and prepare the API
-            const protocol = await this._fetchProtocol(options);
-            api.prepare(this, protocol);
-            // finally connect to the WebSocket
-            await this._connectToWebSocket();
+            if (this.process)
+            {
+                // we first "connect" to stdio pipes, so that we can
+                // first the protocol remotely via the pipe.
+                await this._connect();
+                const protocol = await this._fetchProtocol({});
+                api.prepare(this, protocol);
+            }
+            else
+            {
+                const options = {
+                    host: this.host,
+                    port: this.port,
+                    secure: this.secure,
+                    useHostName: this.useHostName,
+                    alterPath: this.alterPath,
+                    ...this.connectOptions,
+                };
+                // fetch the WebSocket debugger URL
+                const url = await this._fetchWsDebuggerURL(options);
+                // allow the user to alter the URL
+                const urlObject = parseUrl(url);
+                urlObject.pathname = options.alterPath(urlObject.pathname);
+                this.webSocketUrl = formatUrl(urlObject);
+                // update the connection parameters using the debugging URL
+                options.host = urlObject.hostname;
+                options.port = urlObject.port || options.port;
+                // fetch the protocol and prepare the API
+                const protocol = await this._fetchProtocol(options);
+                api.prepare(this, protocol);
+                // finally connect to the WebSocket
+                await this._connect();
+            }
             // since the handler is executed synchronously, the emit() must be
             // performed in the next tick so that uncaught errors in the client code
             // are not intercepted by the Promise mechanism and therefore reported
@@ -167,7 +165,7 @@ class Chrome extends EventEmitter {
     }
 
     // fetch the WebSocket URL according to 'target'
-    async _fetchDebuggerURL(options) {
+    async _fetchWsDebuggerURL(options) {
         const userTarget = this.target;
         switch (typeof userTarget) {
         case 'string': {
@@ -212,40 +210,71 @@ class Chrome extends EventEmitter {
         // otherwise user either the local or the remote version
         else {
             options.local = this.local;
+            if (this.process)
+                options.cdp = this;
             return await devtools.Protocol(options);
         }
     }
 
-    // establish the WebSocket connection and start processing user commands
-    _connectToWebSocket() {
+    _createStdioWrapper() {
+        const stdio = new StdioWrapper(this.process.stdio[3], this.process.stdio[4]);
+        this._close = (...args)=>stdio.close(...args);
+        this._send = (...args)=>stdio.send(...args);
+        return stdio;
+    }
+
+    _createWebSocketWrapper() {
+        if (this.secure) {
+            this.webSocketUrl = this.webSocketUrl.replace(/^ws:/i, 'wss:');
+        }
+        const ws = this._ws = new WebSocket(this.webSocketUrl, [], {
+            followRedirects: true,
+            ...this.connectOptions,
+        });
+        this._close = (callback) => {
+            // don't close if it's already closed
+            if (ws.readyState === 3) {
+                callback();
+            } else {
+                // don't notify on user-initiated shutdown ('disconnect' event)
+                ws.removeAllListeners('close');
+                ws.once('close', () => {
+                    ws.removeAllListeners();
+                    this._handleConnectionClose();
+                    callback();
+                });
+                ws.close();
+            }
+        };
+        this._send = (...args)=>ws.send(...args);
+        return ws;
+    }
+
+    // establish the connection wrapper and start processing user commands
+    _connect() {
         return new Promise((fulfill, reject) => {
-            // create the WebSocket
             try {
-                if (this.secure) {
-                    this.webSocketUrl = this.webSocketUrl.replace(/^ws:/i, 'wss:');
-                }
-                this._ws = new WebSocket(this.webSocketUrl, [], {
-                    followRedirects: true,
-                    ...this.connectOptions
-                });
+                this._transport = this.process
+                    ? this._createStdioWrapper()
+                    : this._createWebSocketWrapper();
             } catch (err) {
-                // handles bad URLs
+                // handle missing stdio streams, bad URLs...
                 reject(err);
                 return;
             }
             // set up event handlers
-            this._ws.on('open', () => {
+            this._transport.on('open', () => {
                 fulfill();
             });
-            this._ws.on('message', (data) => {
+            this._transport.on('message', (data) => {
                 const message = JSON.parse(data);
                 this._handleMessage(message);
             });
-            this._ws.on('close', (code) => {
+            this._transport.on('close', (code) => {
                 this._handleConnectionClose();
                 this.emit('disconnect');
             });
-            this._ws.on('error', (err) => {
+            this._transport.on('error', (err) => {
                 reject(err);
             });
         });
@@ -305,7 +334,7 @@ class Chrome extends EventEmitter {
             sessionId,
             params: params || {}
         };
-        this._ws.send(JSON.stringify(message), (err) => {
+        this._send(JSON.stringify(message), (err) => {
             if (err) {
                 // handle low-level WebSocket errors
                 if (typeof callback === 'function') {
@@ -317,6 +346,7 @@ class Chrome extends EventEmitter {
             }
         });
     }
+    get transport() { return this._transport; }
 }
 
 module.exports = Chrome;

lib/devtools.js

@@ -48,6 +48,13 @@ function promisesWrapper(func) {
 }
 
 function Protocol(options, callback) {
+    // fetch remotely via CDP when using stdio pipes (Bright Data exclusive)
+    if (options.cdp)
+    {
+        return options.cdp.send('Browser.getProtocolJson')
+            .then(data=>callback(null, JSON.parse(data.result)))
+            .catch(callback);
+    }
     // if the local protocol is requested
     if (options.local) {
         const localDescriptor = require('./protocol.json');

lib/stdio-wrapper.js

@@ -0,0 +1,91 @@
+'use strict';
+
+// Adapted from https://github.com/puppeteer/puppeteer/blob/7a2a41f2087b07e8ef1feaf3881bdcc3fd4922ca/src/PipeTransport.js
+
+/**
+ * Copyright 2018 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 { EventEmitter } = require('events');
+
+function addEventListener(emitter, eventName, handler) {
+    emitter.on(eventName, handler);
+    return { emitter, eventName, handler };
+}
+
+function removeEventListeners(listeners) {
+    for (const listener of listeners)
+        listener.emitter.removeListener(listener.eventName, listener.handler);
+    listeners.length = 0;
+}
+
+// wrapper for null-terminated stdio message transport
+class StdioWrapper extends EventEmitter {
+    constructor(pipeWrite, pipeRead) {
+        super();
+        this._pipeWrite = pipeWrite;
+        this._pendingMessage = '';
+        this._eventListeners = [
+            addEventListener(pipeRead, 'data', buffer => this._dispatch(buffer)),
+            addEventListener(pipeRead, 'close', () => {
+                this._pipeWrite = null;
+                this.emit('close');
+            }),
+            addEventListener(pipeRead, 'error', (err) => this.emit('error', err)),
+            addEventListener(pipeWrite, 'error', (err) => this.emit('error', err)),
+        ];
+        setImmediate(() => this.emit('open'));
+    }
+
+    send(message, callback) {
+        try {
+            if (!this._pipeWrite)
+                throw new Error('CDP pipeWrite closed');
+            this._pipeWrite.write(message);
+            this._pipeWrite.write('\0');
+            callback();
+        } catch (err) {
+            callback(err);
+        }
+    }
+
+    _dispatch(buffer) {
+        let end = buffer.indexOf('\0');
+        if (end === -1) {
+            this._pendingMessage += buffer.toString();
+            return;
+        }
+        const message = this._pendingMessage + buffer.toString(undefined, 0, end);
+
+        this.emit('message', message);
+
+        let start = end + 1;
+        end = buffer.indexOf('\0', start);
+        while (end !== -1) {
+            this.emit('message', buffer.toString(undefined, start, end));
+            start = end + 1;
+            end = buffer.indexOf('\0', start);
+        }
+        this._pendingMessage = buffer.toString(undefined, start);
+    }
+
+    close(callback) {
+        this._pipeWrite = null;
+        removeEventListeners(this._eventListeners);
+        callback();
+    }
+}
+
+module.exports = StdioWrapper;