Adding upstream version 1:115.7.0.upstream/1%115.7.0upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

21 files changed, 3100 insertions, 0 deletions

diff --git a/devtools/shared/transport/child-transport.js b/devtools/shared/transport/child-transport.js
new file mode 100644
index 0000000000..52511432e5
--- /dev/null
+++ b/devtools/shared/transport/child-transport.js
@@ -0,0 +1,128 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+const flags = require("resource://devtools/shared/flags.js");
+
+/**
+ * A transport for the debugging protocol that uses nsIMessageManagers to
+ * exchange packets with servers running in child processes.
+ *
+ * In the parent process, |mm| should be the nsIMessageSender for the
+ * child process. In a child process, |mm| should be the child process
+ * message manager, which sends packets to the parent.
+ *
+ * |prefix| is a string included in the message names, to distinguish
+ * multiple servers running in the same child process.
+ *
+ * This transport exchanges messages named 'debug:<prefix>:packet', where
+ * <prefix> is |prefix|, whose data is the protocol packet.
+ */
+function ChildDebuggerTransport(mm, prefix) {
+  this._mm = mm;
+  this._messageName = "debug:" + prefix + ":packet";
+}
+
+/*
+ * To avoid confusion, we use 'message' to mean something that
+ * nsIMessageSender conveys, and 'packet' to mean a remote debugging
+ * protocol packet.
+ */
+ChildDebuggerTransport.prototype = {
+  constructor: ChildDebuggerTransport,
+
+  hooks: null,
+
+  _addListener() {
+    this._mm.addMessageListener(this._messageName, this);
+  },
+
+  _removeListener() {
+    try {
+      this._mm.removeMessageListener(this._messageName, this);
+    } catch (e) {
+      if (e.result != Cr.NS_ERROR_NULL_POINTER) {
+        throw e;
+      }
+      // In some cases, especially when using messageManagers in non-e10s mode, we reach
+      // this point with a dead messageManager which only throws errors but does not
+      // seem to indicate in any other way that it is dead.
+    }
+  },
+
+  ready() {
+    this._addListener();
+  },
+
+  close(options) {
+    this._removeListener();
+    if (this.hooks.onTransportClosed) {
+      this.hooks.onTransportClosed(null, options);
+    }
+  },
+
+  receiveMessage({ data }) {
+    this.hooks.onPacket(data);
+  },
+
+  /**
+   * Helper method to ensure a given `object` can be sent across message manager
+   * without being serialized to JSON.
+   * See https://searchfox.org/mozilla-central/rev/6bfadf95b4a6aaa8bb3b2a166d6c3545983e179a/dom/base/nsFrameMessageManager.cpp#458-469
+   */
+  _canBeSerialized(object) {
+    try {
+      const holder = new StructuredCloneHolder(
+        "ChildDebuggerTransport._canBeSerialized",
+        null,
+        object
+      );
+      holder.deserialize(this);
+    } catch (e) {
+      return false;
+    }
+    return true;
+  },
+
+  pathToUnserializable(object) {
+    for (const key in object) {
+      const value = object[key];
+      if (!this._canBeSerialized(value)) {
+        if (typeof value == "object") {
+          return [key].concat(this.pathToUnserializable(value));
+        }
+        return [key];
+      }
+    }
+    return [];
+  },
+
+  send(packet) {
+    if (flags.testing && !this._canBeSerialized(packet)) {
+      const attributes = this.pathToUnserializable(packet);
+      let msg =
+        "Following packet can't be serialized: " + JSON.stringify(packet);
+      msg += "\nBecause of attributes: " + attributes.join(", ") + "\n";
+      msg += "Did you pass a function or an XPCOM object in it?";
+      throw new Error(msg);
+    }
+    try {
+      this._mm.sendAsyncMessage(this._messageName, packet);
+    } catch (e) {
+      if (e.result != Cr.NS_ERROR_NULL_POINTER) {
+        throw e;
+      }
+      // In some cases, especially when using messageManagers in non-e10s mode, we reach
+      // this point with a dead messageManager which only throws errors but does not
+      // seem to indicate in any other way that it is dead.
+    }
+  },
+
+  startBulkSend() {
+    throw new Error("Can't send bulk data to child processes.");
+  },
+};
+
+exports.ChildDebuggerTransport = ChildDebuggerTransport;
diff --git a/devtools/shared/transport/js-window-actor-transport.js b/devtools/shared/transport/js-window-actor-transport.js
new file mode 100644
index 0000000000..9f0fb82497
--- /dev/null
+++ b/devtools/shared/transport/js-window-actor-transport.js
@@ -0,0 +1,66 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+/**
+ * DevTools transport relying on JS Window Actors. This is an experimental
+ * transport. It is only used when using the JS Window Actor based frame
+ * connector. In that case this transport will be used to communicate between
+ * the DevToolsServer living in the parent process and the DevToolsServer
+ * living in the process of the target frame.
+ *
+ * This is intended to be a replacement for child-transport.js which is a
+ * message-manager based transport.
+ */
+class JsWindowActorTransport {
+  constructor(jsWindowActor, prefix) {
+    this.hooks = null;
+    this._jsWindowActor = jsWindowActor;
+    this._prefix = prefix;
+
+    this._onPacketReceived = this._onPacketReceived.bind(this);
+  }
+
+  _addListener() {
+    this._jsWindowActor.on("packet-received", this._onPacketReceived);
+  }
+
+  _removeListener() {
+    this._jsWindowActor.off("packet-received", this._onPacketReceived);
+  }
+
+  ready() {
+    this._addListener();
+  }
+
+  /**
+   * @param {object} options
+   * @param {boolean} options.isModeSwitching
+   *        true when this is called as the result of a change to the devtools.browsertoolbox.scope pref
+   */
+  close(options) {
+    this._removeListener();
+    if (this.hooks.onTransportClosed) {
+      this.hooks.onTransportClosed(null, options);
+    }
+  }
+
+  _onPacketReceived(eventName, { data }) {
+    const { prefix, packet } = data;
+    if (prefix === this._prefix) {
+      this.hooks.onPacket(packet);
+    }
+  }
+
+  send(packet) {
+    this._jsWindowActor.sendPacket(packet, this._prefix);
+  }
+
+  startBulkSend() {
+    throw new Error("startBulkSend not implemented for JsWindowActorTransport");
+  }
+}
+
+exports.JsWindowActorTransport = JsWindowActorTransport;
diff --git a/devtools/shared/transport/local-transport.js b/devtools/shared/transport/local-transport.js
new file mode 100644
index 0000000000..bff83b7666
--- /dev/null
+++ b/devtools/shared/transport/local-transport.js
@@ -0,0 +1,204 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+const DevToolsUtils = require("resource://devtools/shared/DevToolsUtils.js");
+const { dumpn } = DevToolsUtils;
+const flags = require("resource://devtools/shared/flags.js");
+const StreamUtils = require("resource://devtools/shared/transport/stream-utils.js");
+
+loader.lazyGetter(this, "Pipe", () => {
+  return Components.Constructor("@mozilla.org/pipe;1", "nsIPipe", "init");
+});
+
+/**
+ * An adapter that handles data transfers between the devtools client and
+ * server when they both run in the same process. It presents the same API as
+ * DebuggerTransport, but instead of transmitting serialized messages across a
+ * connection it merely calls the packet dispatcher of the other side.
+ *
+ * @param other LocalDebuggerTransport
+ *        The other endpoint for this debugger connection.
+ *
+ * @see DebuggerTransport
+ */
+function LocalDebuggerTransport(other) {
+  this.other = other;
+  this.hooks = null;
+
+  // A packet number, shared between this and this.other. This isn't used by the
+  // protocol at all, but it makes the packet traces a lot easier to follow.
+  this._serial = this.other ? this.other._serial : { count: 0 };
+  this.close = this.close.bind(this);
+}
+
+LocalDebuggerTransport.prototype = {
+  /**
+   * Transmit a message by directly calling the onPacket handler of the other
+   * endpoint.
+   */
+  send(packet) {
+    const serial = this._serial.count++;
+    if (flags.wantLogging) {
+      // Check 'from' first, as 'echo' packets have both.
+      if (packet.from) {
+        dumpn("Packet " + serial + " sent from " + JSON.stringify(packet.from));
+      } else if (packet.to) {
+        dumpn("Packet " + serial + " sent to " + JSON.stringify(packet.to));
+      }
+    }
+    this._deepFreeze(packet);
+    const other = this.other;
+    if (other) {
+      DevToolsUtils.executeSoon(
+        DevToolsUtils.makeInfallible(() => {
+          // Avoid the cost of JSON.stringify() when logging is disabled.
+          if (flags.wantLogging) {
+            dumpn(
+              "Received packet " +
+                serial +
+                ": " +
+                JSON.stringify(packet, null, 2)
+            );
+          }
+          if (other.hooks) {
+            other.hooks.onPacket(packet);
+          }
+        }, "LocalDebuggerTransport instance's this.other.hooks.onPacket")
+      );
+    }
+  },
+
+  /**
+   * Send a streaming bulk packet directly to the onBulkPacket handler of the
+   * other endpoint.
+   *
+   * This case is much simpler than the full DebuggerTransport, since there is
+   * no primary stream we have to worry about managing while we hand it off to
+   * others temporarily.  Instead, we can just make a single use pipe and be
+   * done with it.
+   */
+  startBulkSend({ actor, type, length }) {
+    const serial = this._serial.count++;
+
+    dumpn("Sent bulk packet " + serial + " for actor " + actor);
+    if (!this.other) {
+      const error = new Error("startBulkSend: other side of transport missing");
+      return Promise.reject(error);
+    }
+
+    const pipe = new Pipe(true, true, 0, 0, null);
+
+    DevToolsUtils.executeSoon(
+      DevToolsUtils.makeInfallible(() => {
+        dumpn("Received bulk packet " + serial);
+        if (!this.other.hooks) {
+          return;
+        }
+
+        // Receiver
+        new Promise(receiverResolve => {
+          const packet = {
+            actor,
+            type,
+            length,
+            copyTo: output => {
+              const copying = StreamUtils.copyStream(
+                pipe.inputStream,
+                output,
+                length
+              );
+              receiverResolve(copying);
+              return copying;
+            },
+            stream: pipe.inputStream,
+            done: receiverResolve,
+          };
+
+          this.other.hooks.onBulkPacket(packet);
+        })
+          // Await the result of reading from the stream
+          .then(() => pipe.inputStream.close(), this.close);
+      }, "LocalDebuggerTransport instance's this.other.hooks.onBulkPacket")
+    );
+
+    // Sender
+    return new Promise(senderResolve => {
+      // The remote transport is not capable of resolving immediately here, so we
+      // shouldn't be able to either.
+      DevToolsUtils.executeSoon(() => {
+        return (
+          new Promise(copyResolve => {
+            senderResolve({
+              copyFrom: input => {
+                const copying = StreamUtils.copyStream(
+                  input,
+                  pipe.outputStream,
+                  length
+                );
+                copyResolve(copying);
+                return copying;
+              },
+              stream: pipe.outputStream,
+              done: copyResolve,
+            });
+          })
+            // Await the result of writing to the stream
+            .then(() => pipe.outputStream.close(), this.close)
+        );
+      });
+    });
+  },
+
+  /**
+   * Close the transport.
+   */
+  close() {
+    if (this.other) {
+      // Remove the reference to the other endpoint before calling close(), to
+      // avoid infinite recursion.
+      const other = this.other;
+      this.other = null;
+      other.close();
+    }
+    if (this.hooks) {
+      try {
+        if (this.hooks.onTransportClosed) {
+          this.hooks.onTransportClosed();
+        }
+      } catch (ex) {
+        console.error(ex);
+      }
+      this.hooks = null;
+    }
+  },
+
+  /**
+   * An empty method for emulating the DebuggerTransport API.
+   */
+  ready() {},
+
+  /**
+   * Helper function that makes an object fully immutable.
+   */
+  _deepFreeze(object) {
+    Object.freeze(object);
+    for (const prop in object) {
+      // Freeze the properties that are objects, not on the prototype, and not
+      // already frozen. Note that this might leave an unfrozen reference
+      // somewhere in the object if there is an already frozen object containing
+      // an unfrozen object.
+      if (
+        object.hasOwnProperty(prop) &&
+        typeof object === "object" &&
+        !Object.isFrozen(object)
+      ) {
+        this._deepFreeze(object[prop]);
+      }
+    }
+  },
+};
+
+exports.LocalDebuggerTransport = LocalDebuggerTransport;
diff --git a/devtools/shared/transport/moz.build b/devtools/shared/transport/moz.build
new file mode 100644
index 0000000000..1ec5d01777
--- /dev/null
+++ b/devtools/shared/transport/moz.build
@@ -0,0 +1,18 @@
+# -*- Mode: python; indent-tabs-mode: nil; tab-width: 40 -*-
+# vim: set filetype=python:
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+XPCSHELL_TESTS_MANIFESTS += ["tests/xpcshell/xpcshell.ini"]
+
+DevToolsModules(
+    "child-transport.js",
+    "js-window-actor-transport.js",
+    "local-transport.js",
+    "packets.js",
+    "stream-utils.js",
+    "transport.js",
+    "websocket-transport.js",
+    "worker-transport.js",
+)
diff --git a/devtools/shared/transport/packets.js b/devtools/shared/transport/packets.js
new file mode 100644
index 0000000000..9f9409a123
--- /dev/null
+++ b/devtools/shared/transport/packets.js
@@ -0,0 +1,440 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+/**
+ * Packets contain read / write functionality for the different packet types
+ * supported by the debugging protocol, so that a transport can focus on
+ * delivery and queue management without worrying too much about the specific
+ * packet types.
+ *
+ * They are intended to be "one use only", so a new packet should be
+ * instantiated for each incoming or outgoing packet.
+ *
+ * A complete Packet type should expose at least the following:
+ *   * read(stream, scriptableStream)
+ *     Called when the input stream has data to read
+ *   * write(stream)
+ *     Called when the output stream is ready to write
+ *   * get done()
+ *     Returns true once the packet is done being read / written
+ *   * destroy()
+ *     Called to clean up at the end of use
+ */
+
+const DevToolsUtils = require("resource://devtools/shared/DevToolsUtils.js");
+const { dumpn, dumpv } = DevToolsUtils;
+const flags = require("resource://devtools/shared/flags.js");
+const StreamUtils = require("resource://devtools/shared/transport/stream-utils.js");
+
+DevToolsUtils.defineLazyGetter(this, "unicodeConverter", () => {
+  // eslint-disable-next-line no-shadow
+  const unicodeConverter = Cc[
+    "@mozilla.org/intl/scriptableunicodeconverter"
+  ].createInstance(Ci.nsIScriptableUnicodeConverter);
+  unicodeConverter.charset = "UTF-8";
+  return unicodeConverter;
+});
+
+// The transport's previous check ensured the header length did not exceed 20
+// characters.  Here, we opt for the somewhat smaller, but still large limit of
+// 1 TiB.
+const PACKET_LENGTH_MAX = Math.pow(2, 40);
+
+/**
+ * A generic Packet processing object (extended by two subtypes below).
+ */
+function Packet(transport) {
+  this._transport = transport;
+  this._length = 0;
+}
+
+/**
+ * Attempt to initialize a new Packet based on the incoming packet header we've
+ * received so far.  We try each of the types in succession, trying JSON packets
+ * first since they are much more common.
+ * @param header string
+ *        The packet header string to attempt parsing.
+ * @param transport DebuggerTransport
+ *        The transport instance that will own the packet.
+ * @return Packet
+ *         The parsed packet of the matching type, or null if no types matched.
+ */
+Packet.fromHeader = function (header, transport) {
+  return (
+    JSONPacket.fromHeader(header, transport) ||
+    BulkPacket.fromHeader(header, transport)
+  );
+};
+
+Packet.prototype = {
+  get length() {
+    return this._length;
+  },
+
+  set length(length) {
+    if (length > PACKET_LENGTH_MAX) {
+      throw Error(
+        "Packet length " +
+          length +
+          " exceeds the max length of " +
+          PACKET_LENGTH_MAX
+      );
+    }
+    this._length = length;
+  },
+
+  destroy() {
+    this._transport = null;
+  },
+};
+
+exports.Packet = Packet;
+
+/**
+ * With a JSON packet (the typical packet type sent via the transport), data is
+ * transferred as a JSON packet serialized into a string, with the string length
+ * prepended to the packet, followed by a colon ([length]:[packet]). The
+ * contents of the JSON packet are specified in the Remote Debugging Protocol
+ * specification.
+ * @param transport DebuggerTransport
+ *        The transport instance that will own the packet.
+ */
+function JSONPacket(transport) {
+  Packet.call(this, transport);
+  this._data = "";
+  this._done = false;
+}
+
+/**
+ * Attempt to initialize a new JSONPacket based on the incoming packet header
+ * we've received so far.
+ * @param header string
+ *        The packet header string to attempt parsing.
+ * @param transport DebuggerTransport
+ *        The transport instance that will own the packet.
+ * @return JSONPacket
+ *         The parsed packet, or null if it's not a match.
+ */
+JSONPacket.fromHeader = function (header, transport) {
+  const match = this.HEADER_PATTERN.exec(header);
+
+  if (!match) {
+    return null;
+  }
+
+  dumpv("Header matches JSON packet");
+  const packet = new JSONPacket(transport);
+  packet.length = +match[1];
+  return packet;
+};
+
+JSONPacket.HEADER_PATTERN = /^(\d+):$/;
+
+JSONPacket.prototype = Object.create(Packet.prototype);
+
+Object.defineProperty(JSONPacket.prototype, "object", {
+  /**
+   * Gets the object (not the serialized string) being read or written.
+   */
+  get() {
+    return this._object;
+  },
+
+  /**
+   * Sets the object to be sent when write() is called.
+   */
+  set(object) {
+    this._object = object;
+    const data = JSON.stringify(object);
+    this._data = unicodeConverter.ConvertFromUnicode(data);
+    this.length = this._data.length;
+  },
+});
+
+JSONPacket.prototype.read = function (stream, scriptableStream) {
+  dumpv("Reading JSON packet");
+
+  // Read in more packet data.
+  this._readData(stream, scriptableStream);
+
+  if (!this.done) {
+    // Don't have a complete packet yet.
+    return;
+  }
+
+  let json = this._data;
+  try {
+    json = unicodeConverter.ConvertToUnicode(json);
+    this._object = JSON.parse(json);
+  } catch (e) {
+    const msg =
+      "Error parsing incoming packet: " +
+      json +
+      " (" +
+      e +
+      " - " +
+      e.stack +
+      ")";
+    console.error(msg);
+    dumpn(msg);
+    return;
+  }
+
+  this._transport._onJSONObjectReady(this._object);
+};
+
+JSONPacket.prototype._readData = function (stream, scriptableStream) {
+  if (flags.wantVerbose) {
+    dumpv(
+      "Reading JSON data: _l: " +
+        this.length +
+        " dL: " +
+        this._data.length +
+        " sA: " +
+        stream.available()
+    );
+  }
+  const bytesToRead = Math.min(
+    this.length - this._data.length,
+    stream.available()
+  );
+  this._data += scriptableStream.readBytes(bytesToRead);
+  this._done = this._data.length === this.length;
+};
+
+JSONPacket.prototype.write = function (stream) {
+  dumpv("Writing JSON packet");
+
+  if (this._outgoing === undefined) {
+    // Format the serialized packet to a buffer
+    this._outgoing = this.length + ":" + this._data;
+  }
+
+  const written = stream.write(this._outgoing, this._outgoing.length);
+  this._outgoing = this._outgoing.slice(written);
+  this._done = !this._outgoing.length;
+};
+
+Object.defineProperty(JSONPacket.prototype, "done", {
+  get() {
+    return this._done;
+  },
+});
+
+JSONPacket.prototype.toString = function () {
+  return JSON.stringify(this._object, null, 2);
+};
+
+exports.JSONPacket = JSONPacket;
+
+/**
+ * With a bulk packet, data is transferred by temporarily handing over the
+ * transport's input or output stream to the application layer for writing data
+ * directly.  This can be much faster for large data sets, and avoids various
+ * stages of copies and data duplication inherent in the JSON packet type.  The
+ * bulk packet looks like:
+ *
+ * bulk [actor] [type] [length]:[data]
+ *
+ * The interpretation of the data portion depends on the kind of actor and the
+ * packet's type.  See the Remote Debugging Protocol Stream Transport spec for
+ * more details.
+ * @param transport DebuggerTransport
+ *        The transport instance that will own the packet.
+ */
+function BulkPacket(transport) {
+  Packet.call(this, transport);
+  this._done = false;
+  let _resolve;
+  this._readyForWriting = new Promise(resolve => {
+    _resolve = resolve;
+  });
+  this._readyForWriting.resolve = _resolve;
+}
+
+/**
+ * Attempt to initialize a new BulkPacket based on the incoming packet header
+ * we've received so far.
+ * @param header string
+ *        The packet header string to attempt parsing.
+ * @param transport DebuggerTransport
+ *        The transport instance that will own the packet.
+ * @return BulkPacket
+ *         The parsed packet, or null if it's not a match.
+ */
+BulkPacket.fromHeader = function (header, transport) {
+  const match = this.HEADER_PATTERN.exec(header);
+
+  if (!match) {
+    return null;
+  }
+
+  dumpv("Header matches bulk packet");
+  const packet = new BulkPacket(transport);
+  packet.header = {
+    actor: match[1],
+    type: match[2],
+    length: +match[3],
+  };
+  return packet;
+};
+
+BulkPacket.HEADER_PATTERN = /^bulk ([^: ]+) ([^: ]+) (\d+):$/;
+
+BulkPacket.prototype = Object.create(Packet.prototype);
+
+BulkPacket.prototype.read = function (stream) {
+  dumpv("Reading bulk packet, handing off input stream");
+
+  // Temporarily pause monitoring of the input stream
+  this._transport.pauseIncoming();
+
+  new Promise(resolve => {
+    this._transport._onBulkReadReady({
+      actor: this.actor,
+      type: this.type,
+      length: this.length,
+      copyTo: output => {
+        dumpv("CT length: " + this.length);
+        const copying = StreamUtils.copyStream(stream, output, this.length);
+        resolve(copying);
+        return copying;
+      },
+      stream,
+      done: resolve,
+    });
+    // Await the result of reading from the stream
+  }).then(() => {
+    dumpv("onReadDone called, ending bulk mode");
+    this._done = true;
+    this._transport.resumeIncoming();
+  }, this._transport.close);
+
+  // Ensure this is only done once
+  this.read = () => {
+    throw new Error("Tried to read() a BulkPacket's stream multiple times.");
+  };
+};
+
+BulkPacket.prototype.write = function (stream) {
+  dumpv("Writing bulk packet");
+
+  if (this._outgoingHeader === undefined) {
+    dumpv("Serializing bulk packet header");
+    // Format the serialized packet header to a buffer
+    this._outgoingHeader =
+      "bulk " + this.actor + " " + this.type + " " + this.length + ":";
+  }
+
+  // Write the header, or whatever's left of it to write.
+  if (this._outgoingHeader.length) {
+    dumpv("Writing bulk packet header");
+    const written = stream.write(
+      this._outgoingHeader,
+      this._outgoingHeader.length
+    );
+    this._outgoingHeader = this._outgoingHeader.slice(written);
+    return;
+  }
+
+  dumpv("Handing off output stream");
+
+  // Temporarily pause the monitoring of the output stream
+  this._transport.pauseOutgoing();
+
+  new Promise(resolve => {
+    this._readyForWriting.resolve({
+      copyFrom: input => {
+        dumpv("CF length: " + this.length);
+        const copying = StreamUtils.copyStream(input, stream, this.length);
+        resolve(copying);
+        return copying;
+      },
+      stream,
+      done: resolve,
+    });
+    // Await the result of writing to the stream
+  }).then(() => {
+    dumpv("onWriteDone called, ending bulk mode");
+    this._done = true;
+    this._transport.resumeOutgoing();
+  }, this._transport.close);
+
+  // Ensure this is only done once
+  this.write = () => {
+    throw new Error("Tried to write() a BulkPacket's stream multiple times.");
+  };
+};
+
+Object.defineProperty(BulkPacket.prototype, "streamReadyForWriting", {
+  get() {
+    return this._readyForWriting;
+  },
+});
+
+Object.defineProperty(BulkPacket.prototype, "header", {
+  get() {
+    return {
+      actor: this.actor,
+      type: this.type,
+      length: this.length,
+    };
+  },
+
+  set(header) {
+    this.actor = header.actor;
+    this.type = header.type;
+    this.length = header.length;
+  },
+});
+
+Object.defineProperty(BulkPacket.prototype, "done", {
+  get() {
+    return this._done;
+  },
+});
+
+BulkPacket.prototype.toString = function () {
+  return "Bulk: " + JSON.stringify(this.header, null, 2);
+};
+
+exports.BulkPacket = BulkPacket;
+
+/**
+ * RawPacket is used to test the transport's error handling of malformed
+ * packets, by writing data directly onto the stream.
+ * @param transport DebuggerTransport
+ *        The transport instance that will own the packet.
+ * @param data string
+ *        The raw string to send out onto the stream.
+ */
+function RawPacket(transport, data) {
+  Packet.call(this, transport);
+  this._data = data;
+  this.length = data.length;
+  this._done = false;
+}
+
+RawPacket.prototype = Object.create(Packet.prototype);
+
+RawPacket.prototype.read = function (stream) {
+  // This hasn't yet been needed for testing.
+  throw Error("Not implmented.");
+};
+
+RawPacket.prototype.write = function (stream) {
+  const written = stream.write(this._data, this._data.length);
+  this._data = this._data.slice(written);
+  this._done = !this._data.length;
+};
+
+Object.defineProperty(RawPacket.prototype, "done", {
+  get() {
+    return this._done;
+  },
+});
+
+exports.RawPacket = RawPacket;
diff --git a/devtools/shared/transport/stream-utils.js b/devtools/shared/transport/stream-utils.js
new file mode 100644
index 0000000000..a3efd1d2aa
--- /dev/null
+++ b/devtools/shared/transport/stream-utils.js
@@ -0,0 +1,254 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+const DevToolsUtils = require("resource://devtools/shared/DevToolsUtils.js");
+const { dumpv } = DevToolsUtils;
+const EventEmitter = require("resource://devtools/shared/event-emitter.js");
+
+DevToolsUtils.defineLazyGetter(this, "IOUtil", () => {
+  return Cc["@mozilla.org/io-util;1"].getService(Ci.nsIIOUtil);
+});
+
+DevToolsUtils.defineLazyGetter(this, "ScriptableInputStream", () => {
+  return Components.Constructor(
+    "@mozilla.org/scriptableinputstream;1",
+    "nsIScriptableInputStream",
+    "init"
+  );
+});
+
+const BUFFER_SIZE = 0x8000;
+
+/**
+ * This helper function (and its companion object) are used by bulk senders and
+ * receivers to read and write data in and out of other streams.  Functions that
+ * make use of this tool are passed to callers when it is time to read or write
+ * bulk data.  It is highly recommended to use these copier functions instead of
+ * the stream directly because the copier enforces the agreed upon length.
+ * Since bulk mode reuses an existing stream, the sender and receiver must write
+ * and read exactly the agreed upon amount of data, or else the entire transport
+ * will be left in a invalid state.  Additionally, other methods of stream
+ * copying (such as NetUtil.asyncCopy) close the streams involved, which would
+ * terminate the debugging transport, and so it is avoided here.
+ *
+ * Overall, this *works*, but clearly the optimal solution would be able to just
+ * use the streams directly.  If it were possible to fully implement
+ * nsIInputStream / nsIOutputStream in JS, wrapper streams could be created to
+ * enforce the length and avoid closing, and consumers could use familiar stream
+ * utilities like NetUtil.asyncCopy.
+ *
+ * The function takes two async streams and copies a precise number of bytes
+ * from one to the other.  Copying begins immediately, but may complete at some
+ * future time depending on data size.  Use the returned promise to know when
+ * it's complete.
+ *
+ * @param input nsIAsyncInputStream
+ *        The stream to copy from.
+ * @param output nsIAsyncOutputStream
+ *        The stream to copy to.
+ * @param length Integer
+ *        The amount of data that needs to be copied.
+ * @return Promise
+ *         The promise is resolved when copying completes or rejected if any
+ *         (unexpected) errors occur.
+ */
+function copyStream(input, output, length) {
+  const copier = new StreamCopier(input, output, length);
+  return copier.copy();
+}
+
+function StreamCopier(input, output, length) {
+  EventEmitter.decorate(this);
+  this._id = StreamCopier._nextId++;
+  this.input = input;
+  // Save off the base output stream, since we know it's async as we've required
+  this.baseAsyncOutput = output;
+  if (IOUtil.outputStreamIsBuffered(output)) {
+    this.output = output;
+  } else {
+    this.output = Cc[
+      "@mozilla.org/network/buffered-output-stream;1"
+    ].createInstance(Ci.nsIBufferedOutputStream);
+    this.output.init(output, BUFFER_SIZE);
+  }
+  this._length = length;
+  this._amountLeft = length;
+  let _resolve;
+  let _reject;
+  this._deferred = new Promise((resolve, reject) => {
+    _resolve = resolve;
+    _reject = reject;
+  });
+  this._deferred.resolve = _resolve;
+  this._deferred.reject = _reject;
+
+  this._copy = this._copy.bind(this);
+  this._flush = this._flush.bind(this);
+  this._destroy = this._destroy.bind(this);
+
+  // Copy promise's then method up to this object.
+  // Allows the copier to offer a promise interface for the simple succeed or
+  // fail scenarios, but also emit events (due to the EventEmitter) for other
+  // states, like progress.
+  this.then = this._deferred.then.bind(this._deferred);
+  this.then(this._destroy, this._destroy);
+
+  // Stream ready callback starts as |_copy|, but may switch to |_flush| at end
+  // if flushing would block the output stream.
+  this._streamReadyCallback = this._copy;
+}
+StreamCopier._nextId = 0;
+
+StreamCopier.prototype = {
+  copy() {
+    // Dispatch to the next tick so that it's possible to attach a progress
+    // event listener, even for extremely fast copies (like when testing).
+    Services.tm.dispatchToMainThread(() => {
+      try {
+        this._copy();
+      } catch (e) {
+        this._deferred.reject(e);
+      }
+    });
+    return this;
+  },
+
+  _copy() {
+    const bytesAvailable = this.input.available();
+    const amountToCopy = Math.min(bytesAvailable, this._amountLeft);
+    this._debug("Trying to copy: " + amountToCopy);
+
+    let bytesCopied;
+    try {
+      bytesCopied = this.output.writeFrom(this.input, amountToCopy);
+    } catch (e) {
+      if (e.result == Cr.NS_BASE_STREAM_WOULD_BLOCK) {
+        this._debug("Base stream would block, will retry");
+        this._debug("Waiting for output stream");
+        this.baseAsyncOutput.asyncWait(this, 0, 0, Services.tm.currentThread);
+        return;
+      }
+      throw e;
+    }
+
+    this._amountLeft -= bytesCopied;
+    this._debug("Copied: " + bytesCopied + ", Left: " + this._amountLeft);
+    this._emitProgress();
+
+    if (this._amountLeft === 0) {
+      this._debug("Copy done!");
+      this._flush();
+      return;
+    }
+
+    this._debug("Waiting for input stream");
+    this.input.asyncWait(this, 0, 0, Services.tm.currentThread);
+  },
+
+  _emitProgress() {
+    this.emit("progress", {
+      bytesSent: this._length - this._amountLeft,
+      totalBytes: this._length,
+    });
+  },
+
+  _flush() {
+    try {
+      this.output.flush();
+    } catch (e) {
+      if (
+        e.result == Cr.NS_BASE_STREAM_WOULD_BLOCK ||
+        e.result == Cr.NS_ERROR_FAILURE
+      ) {
+        this._debug("Flush would block, will retry");
+        this._streamReadyCallback = this._flush;
+        this._debug("Waiting for output stream");
+        this.baseAsyncOutput.asyncWait(this, 0, 0, Services.tm.currentThread);
+        return;
+      }
+      throw e;
+    }
+    this._deferred.resolve();
+  },
+
+  _destroy() {
+    this._destroy = null;
+    this._copy = null;
+    this._flush = null;
+    this.input = null;
+    this.output = null;
+  },
+
+  // nsIInputStreamCallback
+  onInputStreamReady() {
+    this._streamReadyCallback();
+  },
+
+  // nsIOutputStreamCallback
+  onOutputStreamReady() {
+    this._streamReadyCallback();
+  },
+
+  _debug(msg) {
+    // Prefix logs with the copier ID, which makes logs much easier to
+    // understand when several copiers are running simultaneously
+    dumpv("Copier: " + this._id + " " + msg);
+  },
+};
+
+/**
+ * Read from a stream, one byte at a time, up to the next |delimiter|
+ * character, but stopping if we've read |count| without finding it.  Reading
+ * also terminates early if there are less than |count| bytes available on the
+ * stream.  In that case, we only read as many bytes as the stream currently has
+ * to offer.
+ * TODO: This implementation could be removed if bug 984651 is fixed, which
+ *       provides a native version of the same idea.
+ * @param stream nsIInputStream
+ *        The input stream to read from.
+ * @param delimiter string
+ *        The character we're trying to find.
+ * @param count integer
+ *        The max number of characters to read while searching.
+ * @return string
+ *         The data collected.  If the delimiter was found, this string will
+ *         end with it.
+ */
+function delimitedRead(stream, delimiter, count) {
+  dumpv(
+    "Starting delimited read for " + delimiter + " up to " + count + " bytes"
+  );
+
+  let scriptableStream;
+  if (stream instanceof Ci.nsIScriptableInputStream) {
+    scriptableStream = stream;
+  } else {
+    scriptableStream = new ScriptableInputStream(stream);
+  }
+
+  let data = "";
+
+  // Don't exceed what's available on the stream
+  count = Math.min(count, stream.available());
+
+  if (count <= 0) {
+    return data;
+  }
+
+  let char;
+  while (char !== delimiter && count > 0) {
+    char = scriptableStream.readBytes(1);
+    count--;
+    data += char;
+  }
+
+  return data;
+}
+
+module.exports = {
+  copyStream,
+  delimitedRead,
+};
diff --git a/devtools/shared/transport/tests/xpcshell/.eslintrc.js b/devtools/shared/transport/tests/xpcshell/.eslintrc.js
new file mode 100644
index 0000000000..8611c174f5
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/.eslintrc.js
@@ -0,0 +1,6 @@
+"use strict";
+
+module.exports = {
+  // Extend from the common devtools xpcshell eslintrc config.
+  extends: "../../../../.eslintrc.xpcshell.js",
+};
diff --git a/devtools/shared/transport/tests/xpcshell/head_dbg.js b/devtools/shared/transport/tests/xpcshell/head_dbg.js
new file mode 100644
index 0000000000..42794c7712
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/head_dbg.js
@@ -0,0 +1,177 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+
+"use strict";
+
+/* exported Cr, CC, NetUtil, errorCount, initTestDevToolsServer,
+            writeTestTempFile, socket_transport, local_transport, really_long
+*/
+
+var CC = Components.Constructor;
+
+const { require } = ChromeUtils.importESModule(
+  "resource://devtools/shared/loader/Loader.sys.mjs"
+);
+const { NetUtil } = ChromeUtils.import("resource://gre/modules/NetUtil.jsm");
+
+// We do not want to log packets by default, because in some tests,
+// we can be sending large amounts of data. The test harness has
+// trouble dealing with logging all the data, and we end up with
+// intermittent time outs (e.g. bug 775924).
+// Services.prefs.setBoolPref("devtools.debugger.log", true);
+// Services.prefs.setBoolPref("devtools.debugger.log.verbose", true);
+// Enable remote debugging for the relevant tests.
+Services.prefs.setBoolPref("devtools.debugger.remote-enabled", true);
+
+const {
+  ActorRegistry,
+} = require("resource://devtools/server/actors/utils/actor-registry.js");
+const {
+  DevToolsServer,
+} = require("resource://devtools/server/devtools-server.js");
+const {
+  DevToolsClient,
+} = require("resource://devtools/client/devtools-client.js");
+const {
+  SocketListener,
+} = require("resource://devtools/shared/security/socket.js");
+
+// Convert an nsIScriptError 'logLevel' value into an appropriate string.
+function scriptErrorLogLevel(message) {
+  switch (message.logLevel) {
+    case Ci.nsIConsoleMessage.info:
+      return "info";
+    case Ci.nsIConsoleMessage.warn:
+      return "warning";
+    default:
+      Assert.equal(message.logLevel, Ci.nsIConsoleMessage.error);
+      return "error";
+  }
+}
+
+// Register a console listener, so console messages don't just disappear
+// into the ether.
+var errorCount = 0;
+var listener = {
+  observe(message) {
+    errorCount++;
+    let string = "";
+    try {
+      // If we've been given an nsIScriptError, then we can print out
+      // something nicely formatted, for tools like Emacs to pick up.
+      message.QueryInterface(Ci.nsIScriptError);
+      dump(
+        message.sourceName +
+          ":" +
+          message.lineNumber +
+          ": " +
+          scriptErrorLogLevel(message) +
+          ": " +
+          message.errorMessage +
+          "\n"
+      );
+      string = message.errorMessage;
+    } catch (x) {
+      // Be a little paranoid with message, as the whole goal here is to lose
+      // no information.
+      try {
+        string = message.message;
+      } catch (e) {
+        string = "<error converting error message to string>";
+      }
+    }
+
+    // Make sure we exit all nested event loops so that the test can finish.
+    while (DevToolsServer.xpcInspector.eventLoopNestLevel > 0) {
+      DevToolsServer.xpcInspector.exitNestedEventLoop();
+    }
+
+    do_throw("head_dbg.js got console message: " + string + "\n");
+  },
+};
+
+Services.console.registerListener(listener);
+
+/**
+ * Initialize the testing devtools server.
+ */
+function initTestDevToolsServer() {
+  ActorRegistry.registerModule("devtools/server/actors/thread", {
+    prefix: "script",
+    constructor: "ScriptActor",
+    type: { global: true, target: true },
+  });
+  const { createRootActor } = require("xpcshell-test/testactors");
+  DevToolsServer.setRootActor(createRootActor);
+  // Allow incoming connections.
+  DevToolsServer.init();
+  // Avoid the server from being destroyed when the last connection closes
+  DevToolsServer.keepAlive = true;
+}
+
+/**
+ * Wrapper around do_get_file to prefix files with the name of current test to
+ * avoid collisions when running in parallel.
+ */
+function getTestTempFile(fileName, allowMissing) {
+  let thisTest = _TEST_FILE.toString().replace(/\\/g, "/");
+  thisTest = thisTest.substring(thisTest.lastIndexOf("/") + 1);
+  thisTest = thisTest.replace(/\..*$/, "");
+  return do_get_file(fileName + "-" + thisTest, allowMissing);
+}
+
+function writeTestTempFile(fileName, content) {
+  const file = getTestTempFile(fileName, true);
+  const stream = Cc["@mozilla.org/network/file-output-stream;1"].createInstance(
+    Ci.nsIFileOutputStream
+  );
+  stream.init(file, -1, -1, 0);
+  try {
+    do {
+      const numWritten = stream.write(content, content.length);
+      content = content.slice(numWritten);
+    } while (content.length);
+  } finally {
+    stream.close();
+  }
+}
+
+/** * Transport Factories ***/
+
+var socket_transport = async function () {
+  if (!DevToolsServer.listeningSockets) {
+    const AuthenticatorType = DevToolsServer.Authenticators.get("PROMPT");
+    const authenticator = new AuthenticatorType.Server();
+    authenticator.allowConnection = () => {
+      return DevToolsServer.AuthenticationResult.ALLOW;
+    };
+    const socketOptions = {
+      authenticator,
+      portOrPath: -1,
+    };
+    const debuggerListener = new SocketListener(DevToolsServer, socketOptions);
+    await debuggerListener.open();
+  }
+  const port = DevToolsServer._listeners[0].port;
+  info("DevTools server port is " + port);
+  return DevToolsClient.socketConnect({ host: "127.0.0.1", port });
+};
+
+function local_transport() {
+  return Promise.resolve(DevToolsServer.connectPipe());
+}
+
+/** * Sample Data ***/
+
+var gReallyLong;
+function really_long() {
+  if (gReallyLong) {
+    return gReallyLong;
+  }
+  let ret = "0123456789";
+  for (let i = 0; i < 18; i++) {
+    ret += ret;
+  }
+  gReallyLong = ret;
+  return ret;
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_bulk_error.js b/devtools/shared/transport/tests/xpcshell/test_bulk_error.js
new file mode 100644
index 0000000000..52cc826e51
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_bulk_error.js
@@ -0,0 +1,94 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+
+"use strict";
+
+function run_test() {
+  initTestDevToolsServer();
+  add_test_bulk_actor();
+
+  add_task(async function () {
+    await test_string_error(socket_transport, json_reply);
+    await test_string_error(local_transport, json_reply);
+    DevToolsServer.destroy();
+  });
+
+  run_next_test();
+}
+
+/** * Sample Bulk Actor ***/
+const { Actor } = require("resource://devtools/shared/protocol/Actor.js");
+class TestBulkActor extends Actor {
+  constructor(conn) {
+    super(conn);
+
+    this.typeName = "testBulk";
+    this.requestTypes = {
+      jsonReply: this.jsonReply,
+    };
+  }
+
+  jsonReply({ length, reader, reply, done }) {
+    Assert.equal(length, really_long().length);
+
+    return {
+      allDone: true,
+    };
+  }
+}
+
+function add_test_bulk_actor() {
+  ActorRegistry.addGlobalActor(
+    {
+      constructorName: "TestBulkActor",
+      constructorFun: TestBulkActor,
+    },
+    "testBulk"
+  );
+}
+
+/** * Tests ***/
+
+var test_string_error = async function (transportFactory, onReady) {
+  const transport = await transportFactory();
+
+  const client = new DevToolsClient(transport);
+  await client.connect();
+  const response = await client.mainRoot.rootForm;
+
+  await onReady(client, response);
+  client.close();
+  transport.close();
+};
+
+/** * Reply Types ***/
+
+function json_reply(client, response) {
+  const reallyLong = really_long();
+
+  const request = client.startBulkRequest({
+    actor: response.testBulk,
+    type: "jsonReply",
+    length: reallyLong.length,
+  });
+
+  // Send bulk data to server
+  return new Promise(resolve => {
+    request.on("bulk-send-ready", ({ writer, done }) => {
+      const input = Cc["@mozilla.org/io/string-input-stream;1"].createInstance(
+        Ci.nsIStringInputStream
+      );
+      input.setData(reallyLong, reallyLong.length);
+      try {
+        writer.copyFrom(input, () => {
+          input.close();
+          done();
+        });
+        do_throw(new Error("Copying should fail, the stream is not async."));
+      } catch (e) {
+        Assert.ok(true);
+        resolve();
+      }
+    });
+  });
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_client_server_bulk.js b/devtools/shared/transport/tests/xpcshell/test_client_server_bulk.js
new file mode 100644
index 0000000000..22efdc6eba
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_client_server_bulk.js
@@ -0,0 +1,312 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+
+"use strict";
+
+var { FileUtils } = ChromeUtils.importESModule(
+  "resource://gre/modules/FileUtils.sys.mjs"
+);
+var Pipe = Components.Constructor("@mozilla.org/pipe;1", "nsIPipe", "init");
+
+function run_test() {
+  initTestDevToolsServer();
+  add_test_bulk_actor();
+
+  add_task(async function () {
+    await test_bulk_request_cs(socket_transport, "jsonReply", "json");
+    await test_bulk_request_cs(local_transport, "jsonReply", "json");
+    await test_bulk_request_cs(socket_transport, "bulkEcho", "bulk");
+    await test_bulk_request_cs(local_transport, "bulkEcho", "bulk");
+    await test_json_request_cs(socket_transport, "bulkReply", "bulk");
+    await test_json_request_cs(local_transport, "bulkReply", "bulk");
+    DevToolsServer.destroy();
+  });
+
+  run_next_test();
+}
+
+/** * Sample Bulk Actor ***/
+const { Actor } = require("resource://devtools/shared/protocol/Actor.js");
+class TestBulkActor extends Actor {
+  constructor(conn) {
+    super(conn, { typeName: "testBulk", methods: [] });
+
+    this.requestTypes = {
+      bulkEcho: this.bulkEcho,
+      bulkReply: this.bulkReply,
+      jsonReply: this.jsonReply,
+    };
+  }
+
+  bulkEcho({ actor, type, length, copyTo }) {
+    Assert.equal(length, really_long().length);
+    this.conn
+      .startBulkSend({
+        actor,
+        type,
+        length,
+      })
+      .then(({ copyFrom }) => {
+        // We'll just echo back the same thing
+        const pipe = new Pipe(true, true, 0, 0, null);
+        copyTo(pipe.outputStream).then(() => {
+          pipe.outputStream.close();
+        });
+        copyFrom(pipe.inputStream).then(() => {
+          pipe.inputStream.close();
+        });
+      });
+  }
+
+  bulkReply({ to, type }) {
+    this.conn
+      .startBulkSend({
+        actor: to,
+        type,
+        length: really_long().length,
+      })
+      .then(({ copyFrom }) => {
+        NetUtil.asyncFetch(
+          {
+            uri: NetUtil.newURI(getTestTempFile("bulk-input")),
+            loadUsingSystemPrincipal: true,
+          },
+          input => {
+            copyFrom(input).then(() => {
+              input.close();
+            });
+          }
+        );
+      });
+  }
+
+  jsonReply({ length, copyTo }) {
+    Assert.equal(length, really_long().length);
+
+    const outputFile = getTestTempFile("bulk-output", true);
+    outputFile.create(Ci.nsIFile.NORMAL_FILE_TYPE, parseInt("666", 8));
+
+    const output = FileUtils.openSafeFileOutputStream(outputFile);
+
+    return copyTo(output)
+      .then(() => {
+        FileUtils.closeSafeFileOutputStream(output);
+        return verify_files();
+      })
+      .then(() => {
+        return { allDone: true };
+      }, do_throw);
+  }
+}
+
+function add_test_bulk_actor() {
+  ActorRegistry.addGlobalActor(
+    {
+      constructorName: "TestBulkActor",
+      constructorFun: TestBulkActor,
+    },
+    "testBulk"
+  );
+}
+
+/** * Reply Handlers ***/
+
+var replyHandlers = {
+  json(request) {
+    // Receive JSON reply from server
+    return new Promise(resolve => {
+      request.on("json-reply", reply => {
+        Assert.ok(reply.allDone);
+        resolve();
+      });
+    });
+  },
+
+  bulk(request) {
+    // Receive bulk data reply from server
+    return new Promise(resolve => {
+      request.on("bulk-reply", ({ length, copyTo }) => {
+        Assert.equal(length, really_long().length);
+
+        const outputFile = getTestTempFile("bulk-output", true);
+        outputFile.create(Ci.nsIFile.NORMAL_FILE_TYPE, parseInt("666", 8));
+
+        const output = FileUtils.openSafeFileOutputStream(outputFile);
+
+        copyTo(output).then(() => {
+          FileUtils.closeSafeFileOutputStream(output);
+          resolve(verify_files());
+        });
+      });
+    });
+  },
+};
+
+/** * Tests ***/
+
+var test_bulk_request_cs = async function (
+  transportFactory,
+  actorType,
+  replyType
+) {
+  // Ensure test files are not present from a failed run
+  cleanup_files();
+  writeTestTempFile("bulk-input", really_long());
+
+  let clientResolve;
+  const clientDeferred = new Promise(resolve => {
+    clientResolve = resolve;
+  });
+
+  let serverResolve;
+  const serverDeferred = new Promise(resolve => {
+    serverResolve = resolve;
+  });
+
+  let bulkCopyResolve;
+  const bulkCopyDeferred = new Promise(resolve => {
+    bulkCopyResolve = resolve;
+  });
+
+  const transport = await transportFactory();
+
+  const client = new DevToolsClient(transport);
+  client.connect().then(() => {
+    client.mainRoot.rootForm.then(clientResolve);
+  });
+
+  function bulkSendReadyCallback({ copyFrom }) {
+    NetUtil.asyncFetch(
+      {
+        uri: NetUtil.newURI(getTestTempFile("bulk-input")),
+        loadUsingSystemPrincipal: true,
+      },
+      input => {
+        copyFrom(input).then(() => {
+          input.close();
+          bulkCopyResolve();
+        });
+      }
+    );
+  }
+
+  clientDeferred
+    .then(response => {
+      const request = client.startBulkRequest({
+        actor: response.testBulk,
+        type: actorType,
+        length: really_long().length,
+      });
+
+      // Send bulk data to server
+      request.on("bulk-send-ready", bulkSendReadyCallback);
+
+      // Set up reply handling for this type
+      replyHandlers[replyType](request).then(() => {
+        client.close();
+        transport.close();
+      });
+    })
+    .catch(do_throw);
+
+  DevToolsServer.on("connectionchange", type => {
+    if (type === "closed") {
+      serverResolve();
+    }
+  });
+
+  return Promise.all([clientDeferred, bulkCopyDeferred, serverDeferred]);
+};
+
+var test_json_request_cs = async function (
+  transportFactory,
+  actorType,
+  replyType
+) {
+  // Ensure test files are not present from a failed run
+  cleanup_files();
+  writeTestTempFile("bulk-input", really_long());
+
+  let clientResolve;
+  const clientDeferred = new Promise(resolve => {
+    clientResolve = resolve;
+  });
+
+  let serverResolve;
+  const serverDeferred = new Promise(resolve => {
+    serverResolve = resolve;
+  });
+
+  const transport = await transportFactory();
+
+  const client = new DevToolsClient(transport);
+  await client.connect();
+  client.mainRoot.rootForm.then(clientResolve);
+
+  clientDeferred
+    .then(response => {
+      const request = client.request({
+        to: response.testBulk,
+        type: actorType,
+      });
+
+      // Set up reply handling for this type
+      replyHandlers[replyType](request).then(() => {
+        client.close();
+        transport.close();
+      });
+    })
+    .catch(do_throw);
+
+  DevToolsServer.on("connectionchange", type => {
+    if (type === "closed") {
+      serverResolve();
+    }
+  });
+
+  return Promise.all([clientDeferred, serverDeferred]);
+};
+
+/** * Test Utils ***/
+
+function verify_files() {
+  const reallyLong = really_long();
+
+  const inputFile = getTestTempFile("bulk-input");
+  const outputFile = getTestTempFile("bulk-output");
+
+  Assert.equal(inputFile.fileSize, reallyLong.length);
+  Assert.equal(outputFile.fileSize, reallyLong.length);
+
+  // Ensure output file contents actually match
+  return new Promise(resolve => {
+    NetUtil.asyncFetch(
+      {
+        uri: NetUtil.newURI(getTestTempFile("bulk-output")),
+        loadUsingSystemPrincipal: true,
+      },
+      input => {
+        const outputData = NetUtil.readInputStreamToString(
+          input,
+          reallyLong.length
+        );
+        // Avoid do_check_eq here so we don't log the contents
+        Assert.ok(outputData === reallyLong);
+        input.close();
+        resolve();
+      }
+    );
+  }).then(cleanup_files);
+}
+
+function cleanup_files() {
+  const inputFile = getTestTempFile("bulk-input", true);
+  if (inputFile.exists()) {
+    inputFile.remove(false);
+  }
+
+  const outputFile = getTestTempFile("bulk-output", true);
+  if (outputFile.exists()) {
+    outputFile.remove(false);
+  }
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_dbgsocket.js b/devtools/shared/transport/tests/xpcshell/test_dbgsocket.js
new file mode 100644
index 0000000000..535431aa38
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_dbgsocket.js
@@ -0,0 +1,163 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+"use strict";
+
+/* global structuredClone */
+
+var gPort;
+var gExtraListener;
+
+function run_test() {
+  info("Starting test at " + new Date().toTimeString());
+  initTestDevToolsServer();
+
+  add_task(test_socket_conn);
+  add_task(test_socket_shutdown);
+  add_test(test_pipe_conn);
+
+  run_next_test();
+}
+
+const { Actor } = require("resource://devtools/shared/protocol/Actor.js");
+class EchoTestActor extends Actor {
+  constructor(conn) {
+    super(conn, { typeName: "EchoTestActor", methods: [] });
+
+    this.requestTypes = {
+      echo: EchoTestActor.prototype.onEcho,
+    };
+  }
+
+  onEcho(request) {
+    /*
+     * Request packets are frozen. Copy request, so that
+     * DevToolsServerConnection.onPacket can attach a 'from' property.
+     */
+    return structuredClone(request);
+  }
+}
+
+async function test_socket_conn() {
+  Assert.equal(DevToolsServer.listeningSockets, 0);
+  const AuthenticatorType = DevToolsServer.Authenticators.get("PROMPT");
+  const authenticator = new AuthenticatorType.Server();
+  authenticator.allowConnection = () => {
+    return DevToolsServer.AuthenticationResult.ALLOW;
+  };
+  const socketOptions = {
+    authenticator,
+    portOrPath: -1,
+  };
+  const listener = new SocketListener(DevToolsServer, socketOptions);
+  Assert.ok(listener);
+  listener.open();
+  Assert.equal(DevToolsServer.listeningSockets, 1);
+  gPort = DevToolsServer._listeners[0].port;
+  info("DevTools server port is " + gPort);
+  // Open a second, separate listener
+  gExtraListener = new SocketListener(DevToolsServer, socketOptions);
+  gExtraListener.open();
+  Assert.equal(DevToolsServer.listeningSockets, 2);
+  Assert.ok(!DevToolsServer.hasConnection());
+
+  info("Starting long and unicode tests at " + new Date().toTimeString());
+  // We can't use EventEmitter.once as this is the second argument we care about...
+  const onConnectionChange = new Promise(res => {
+    DevToolsServer.once("connectionchange", (type, conn) => res(conn));
+  });
+
+  const transport = await DevToolsClient.socketConnect({
+    host: "127.0.0.1",
+    port: gPort,
+  });
+  Assert.ok(DevToolsServer.hasConnection());
+  info("Wait for server connection");
+  const conn = await onConnectionChange;
+
+  // Register a custom actor to do echo requests
+  const actor = new EchoTestActor(conn);
+  actor.actorID = "echo-actor";
+  conn.addActor(actor);
+
+  // Assert that connection settings are available on transport object
+  const settings = transport.connectionSettings;
+  Assert.equal(settings.host, "127.0.0.1");
+  Assert.equal(settings.port, gPort);
+
+  const onDebuggerConnectionClosed = DevToolsServer.once("connectionchange");
+  const unicodeString = "(╯°□°）╯︵ ┻━┻";
+  await new Promise(resolve => {
+    transport.hooks = {
+      onPacket(packet) {
+        this.onPacket = function ({ unicode }) {
+          Assert.equal(unicode, unicodeString);
+          transport.close();
+        };
+        // Verify that things work correctly when bigger than the output
+        // transport buffers and when transporting unicode...
+        transport.send({
+          to: "echo-actor",
+          type: "echo",
+          reallylong: really_long(),
+          unicode: unicodeString,
+        });
+        Assert.equal(packet.from, "root");
+      },
+      onTransportClosed(status) {
+        resolve();
+      },
+    };
+    transport.ready();
+  });
+  const type = await onDebuggerConnectionClosed;
+  Assert.equal(type, "closed");
+  Assert.ok(!DevToolsServer.hasConnection());
+}
+
+async function test_socket_shutdown() {
+  Assert.equal(DevToolsServer.listeningSockets, 2);
+  gExtraListener.close();
+  Assert.equal(DevToolsServer.listeningSockets, 1);
+  Assert.ok(DevToolsServer.closeAllSocketListeners());
+  Assert.equal(DevToolsServer.listeningSockets, 0);
+  // Make sure closing the listener twice does nothing.
+  Assert.ok(!DevToolsServer.closeAllSocketListeners());
+  Assert.equal(DevToolsServer.listeningSockets, 0);
+
+  info("Connecting to a server socket at " + new Date().toTimeString());
+  try {
+    await DevToolsClient.socketConnect({
+      host: "127.0.0.1",
+      port: gPort,
+    });
+  } catch (e) {
+    if (
+      e.result == Cr.NS_ERROR_CONNECTION_REFUSED ||
+      e.result == Cr.NS_ERROR_NET_TIMEOUT
+    ) {
+      // The connection should be refused here, but on slow or overloaded
+      // machines it may just time out.
+      Assert.ok(true);
+      return;
+    }
+    throw e;
+  }
+
+  // Shouldn't reach this, should never connect.
+  Assert.ok(false);
+}
+
+function test_pipe_conn() {
+  const transport = DevToolsServer.connectPipe();
+  transport.hooks = {
+    onPacket(packet) {
+      Assert.equal(packet.from, "root");
+      transport.close();
+    },
+    onTransportClosed(status) {
+      run_next_test();
+    },
+  };
+
+  transport.ready();
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_dbgsocket_connection_drop.js b/devtools/shared/transport/tests/xpcshell/test_dbgsocket_connection_drop.js
new file mode 100644
index 0000000000..e08c2380fb
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_dbgsocket_connection_drop.js
@@ -0,0 +1,86 @@
+/**
+ * Any copyright is dedicated to the Public Domain.
+ * http://creativecommons.org/publicdomain/zero/1.0/
+ */
+
+/**
+ * Bug 755412 - checks if the server drops the connection on an improperly
+ * framed packet, i.e. when the length header is invalid.
+ */
+"use strict";
+
+const {
+  RawPacket,
+} = require("resource://devtools/shared/transport/packets.js");
+
+function run_test() {
+  info("Starting test at " + new Date().toTimeString());
+  initTestDevToolsServer();
+
+  add_task(test_socket_conn_drops_after_invalid_header);
+  add_task(test_socket_conn_drops_after_invalid_header_2);
+  add_task(test_socket_conn_drops_after_too_large_length);
+  add_task(test_socket_conn_drops_after_too_long_header);
+  run_next_test();
+}
+
+function test_socket_conn_drops_after_invalid_header() {
+  return test_helper('fluff30:27:{"to":"root","type":"echo"}');
+}
+
+function test_socket_conn_drops_after_invalid_header_2() {
+  return test_helper('27asd:{"to":"root","type":"echo"}');
+}
+
+function test_socket_conn_drops_after_too_large_length() {
+  // Packet length is limited (semi-arbitrarily) to 1 TiB (2^40)
+  return test_helper("4305724038957487634549823475894325:");
+}
+
+function test_socket_conn_drops_after_too_long_header() {
+  // The packet header is currently limited to no more than 200 bytes
+  let rawPacket = "4305724038957487634549823475894325";
+  for (let i = 0; i < 8; i++) {
+    rawPacket += rawPacket;
+  }
+  return test_helper(rawPacket + ":");
+}
+
+var test_helper = async function (payload) {
+  const AuthenticatorType = DevToolsServer.Authenticators.get("PROMPT");
+  const authenticator = new AuthenticatorType.Server();
+  authenticator.allowConnection = () => {
+    return DevToolsServer.AuthenticationResult.ALLOW;
+  };
+  const socketOptions = {
+    authenticator,
+    portOrPath: -1,
+  };
+
+  const listener = new SocketListener(DevToolsServer, socketOptions);
+  listener.open();
+
+  const transport = await DevToolsClient.socketConnect({
+    host: "127.0.0.1",
+    port: listener.port,
+  });
+  return new Promise(resolve => {
+    transport.hooks = {
+      onPacket(packet) {
+        this.onPacket = function () {
+          do_throw(new Error("This connection should be dropped."));
+          transport.close();
+        };
+
+        // Inject the payload directly into the stream.
+        transport._outgoing.push(new RawPacket(transport, payload));
+        transport._flushOutgoing();
+      },
+      onTransportClosed(status) {
+        Assert.ok(true);
+        resolve();
+      },
+    };
+    transport.ready();
+  });
+};
diff --git a/devtools/shared/transport/tests/xpcshell/test_delimited_read.js b/devtools/shared/transport/tests/xpcshell/test_delimited_read.js
new file mode 100644
index 0000000000..fe94f81bbf
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_delimited_read.js
@@ -0,0 +1,30 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+"use strict";
+
+const StreamUtils = require("resource://devtools/shared/transport/stream-utils.js");
+
+const StringInputStream = Components.Constructor(
+  "@mozilla.org/io/string-input-stream;1",
+  "nsIStringInputStream",
+  "setData"
+);
+
+function run_test() {
+  add_task(async function () {
+    await test_delimited_read("0123:", "0123:");
+    await test_delimited_read("0123:4567:", "0123:");
+    await test_delimited_read("012345678901:", "0123456789");
+    await test_delimited_read("0123/0123", "0123/0123");
+  });
+
+  run_next_test();
+}
+
+/** * Tests ***/
+
+function test_delimited_read(input, expected) {
+  input = new StringInputStream(input, input.length);
+  const result = StreamUtils.delimitedRead(input, ":", 10);
+  Assert.equal(result, expected);
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_packet.js b/devtools/shared/transport/tests/xpcshell/test_packet.js
new file mode 100644
index 0000000000..459a7a8211
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_packet.js
@@ -0,0 +1,24 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+"use strict";
+
+const {
+  JSONPacket,
+  BulkPacket,
+} = require("resource://devtools/shared/transport/packets.js");
+
+function run_test() {
+  add_test(test_packet_done);
+  run_next_test();
+}
+
+// Ensure done can be checked without getting an error
+function test_packet_done() {
+  const json = new JSONPacket();
+  Assert.ok(!json.done);
+
+  const bulk = new BulkPacket();
+  Assert.ok(!bulk.done);
+
+  run_next_test();
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_queue.js b/devtools/shared/transport/tests/xpcshell/test_queue.js
new file mode 100644
index 0000000000..603640e34d
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_queue.js
@@ -0,0 +1,198 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+
+"use strict";
+
+/**
+ * This test verifies that the transport's queue operates correctly when various
+ * packets are scheduled simultaneously.
+ */
+
+var { FileUtils } = ChromeUtils.importESModule(
+  "resource://gre/modules/FileUtils.sys.mjs"
+);
+
+function run_test() {
+  initTestDevToolsServer();
+
+  add_task(async function () {
+    await test_transport(socket_transport);
+    await test_transport(local_transport);
+    DevToolsServer.destroy();
+  });
+
+  run_next_test();
+}
+
+/** * Tests ***/
+
+var test_transport = async function (transportFactory) {
+  let clientResolve;
+  const clientDeferred = new Promise(resolve => {
+    clientResolve = resolve;
+  });
+
+  let serverResolve;
+  const serverDeferred = new Promise(resolve => {
+    serverResolve = resolve;
+  });
+
+  // Ensure test files are not present from a failed run
+  cleanup_files();
+  const reallyLong = really_long();
+  writeTestTempFile("bulk-input", reallyLong);
+
+  Assert.equal(Object.keys(DevToolsServer._connections).length, 0);
+
+  const transport = await transportFactory();
+
+  // Sending from client to server
+  function write_data({ copyFrom }) {
+    NetUtil.asyncFetch(
+      {
+        uri: NetUtil.newURI(getTestTempFile("bulk-input")),
+        loadUsingSystemPrincipal: true,
+      },
+      function (input, status) {
+        copyFrom(input).then(() => {
+          input.close();
+        });
+      }
+    );
+  }
+
+  // Receiving on server from client
+  function on_bulk_packet({ actor, type, length, copyTo }) {
+    Assert.equal(actor, "root");
+    Assert.equal(type, "file-stream");
+    Assert.equal(length, reallyLong.length);
+
+    const outputFile = getTestTempFile("bulk-output", true);
+    outputFile.create(Ci.nsIFile.NORMAL_FILE_TYPE, parseInt("666", 8));
+
+    const output = FileUtils.openSafeFileOutputStream(outputFile);
+
+    copyTo(output)
+      .then(() => {
+        FileUtils.closeSafeFileOutputStream(output);
+        return verify();
+      })
+      .then(() => {
+        // It's now safe to close
+        transport.hooks.onTransportClosed = () => {
+          clientResolve();
+        };
+        transport.close();
+      });
+  }
+
+  // Client
+
+  function send_packets() {
+    // Specifically, we want to ensure that multiple send()s proceed without
+    // causing the transport to die.
+    transport.send({
+      actor: "root",
+      type: "explode",
+    });
+
+    transport
+      .startBulkSend({
+        actor: "root",
+        type: "file-stream",
+        length: reallyLong.length,
+      })
+      .then(write_data);
+  }
+
+  transport.hooks = {
+    onPacket(packet) {
+      if (packet.error) {
+        transport.hooks.onError(packet);
+      } else if (packet.applicationType) {
+        transport.hooks.onServerHello(packet);
+      } else {
+        do_throw("Unexpected server reply");
+      }
+    },
+
+    onServerHello(packet) {
+      // We've received the initial start up packet
+      Assert.equal(packet.from, "root");
+      Assert.equal(packet.applicationType, "xpcshell-tests");
+
+      // Server
+      Assert.equal(Object.keys(DevToolsServer._connections).length, 1);
+      info(Object.keys(DevToolsServer._connections));
+      for (const connId in DevToolsServer._connections) {
+        DevToolsServer._connections[connId].onBulkPacket = on_bulk_packet;
+      }
+
+      DevToolsServer.on("connectionchange", type => {
+        if (type === "closed") {
+          serverResolve();
+        }
+      });
+
+      send_packets();
+    },
+
+    onError(packet) {
+      // The explode actor doesn't exist
+      Assert.equal(packet.from, "root");
+      Assert.equal(packet.error, "noSuchActor");
+    },
+
+    onTransportClosed() {
+      do_throw("Transport closed before we expected");
+    },
+  };
+
+  transport.ready();
+
+  return Promise.all([clientDeferred, serverDeferred]);
+};
+
+/** * Test Utils ***/
+
+function verify() {
+  const reallyLong = really_long();
+
+  const inputFile = getTestTempFile("bulk-input");
+  const outputFile = getTestTempFile("bulk-output");
+
+  Assert.equal(inputFile.fileSize, reallyLong.length);
+  Assert.equal(outputFile.fileSize, reallyLong.length);
+
+  // Ensure output file contents actually match
+  return new Promise(resolve => {
+    NetUtil.asyncFetch(
+      {
+        uri: NetUtil.newURI(getTestTempFile("bulk-output")),
+        loadUsingSystemPrincipal: true,
+      },
+      input => {
+        const outputData = NetUtil.readInputStreamToString(
+          input,
+          reallyLong.length
+        );
+        // Avoid do_check_eq here so we don't log the contents
+        Assert.ok(outputData === reallyLong);
+        input.close();
+        resolve();
+      }
+    );
+  }).then(cleanup_files);
+}
+
+function cleanup_files() {
+  const inputFile = getTestTempFile("bulk-input", true);
+  if (inputFile.exists()) {
+    inputFile.remove(false);
+  }
+
+  const outputFile = getTestTempFile("bulk-output", true);
+  if (outputFile.exists()) {
+    outputFile.remove(false);
+  }
+}
diff --git a/devtools/shared/transport/tests/xpcshell/test_transport_bulk.js b/devtools/shared/transport/tests/xpcshell/test_transport_bulk.js
new file mode 100644
index 0000000000..137cbd2679
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/test_transport_bulk.js
@@ -0,0 +1,169 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+
+"use strict";
+
+var { FileUtils } = ChromeUtils.importESModule(
+  "resource://gre/modules/FileUtils.sys.mjs"
+);
+
+function run_test() {
+  initTestDevToolsServer();
+
+  add_task(async function () {
+    await test_bulk_transfer_transport(socket_transport);
+    await test_bulk_transfer_transport(local_transport);
+    DevToolsServer.destroy();
+  });
+
+  run_next_test();
+}
+
+/** * Tests ***/
+
+/**
+ * This tests a one-way bulk transfer at the transport layer.
+ */
+var test_bulk_transfer_transport = async function (transportFactory) {
+  info("Starting bulk transfer test at " + new Date().toTimeString());
+
+  let clientResolve;
+  const clientDeferred = new Promise(resolve => {
+    clientResolve = resolve;
+  });
+
+  let serverResolve;
+  const serverDeferred = new Promise(resolve => {
+    serverResolve = resolve;
+  });
+
+  // Ensure test files are not present from a failed run
+  cleanup_files();
+  const reallyLong = really_long();
+  writeTestTempFile("bulk-input", reallyLong);
+
+  Assert.equal(Object.keys(DevToolsServer._connections).length, 0);
+
+  const transport = await transportFactory();
+
+  // Sending from client to server
+  function write_data({ copyFrom }) {
+    NetUtil.asyncFetch(
+      {
+        uri: NetUtil.newURI(getTestTempFile("bulk-input")),
+        loadUsingSystemPrincipal: true,
+      },
+      function (input, status) {
+        copyFrom(input).then(() => {
+          input.close();
+        });
+      }
+    );
+  }
+
+  // Receiving on server from client
+  function on_bulk_packet({ actor, type, length, copyTo }) {
+    Assert.equal(actor, "root");
+    Assert.equal(type, "file-stream");
+    Assert.equal(length, reallyLong.length);
+
+    const outputFile = getTestTempFile("bulk-output", true);
+    outputFile.create(Ci.nsIFile.NORMAL_FILE_TYPE, parseInt("666", 8));
+
+    const output = FileUtils.openSafeFileOutputStream(outputFile);
+
+    copyTo(output)
+      .then(() => {
+        FileUtils.closeSafeFileOutputStream(output);
+        return verify();
+      })
+      .then(() => {
+        // It's now safe to close
+        transport.hooks.onTransportClosed = () => {
+          clientResolve();
+        };
+        transport.close();
+      });
+  }
+
+  // Client
+  transport.hooks = {
+    onPacket(packet) {
+      // We've received the initial start up packet
+      Assert.equal(packet.from, "root");
+
+      // Server
+      Assert.equal(Object.keys(DevToolsServer._connections).length, 1);
+      info(Object.keys(DevToolsServer._connections));
+      for (const connId in DevToolsServer._connections) {
+        DevToolsServer._connections[connId].onBulkPacket = on_bulk_packet;
+      }
+
+      DevToolsServer.on("connectionchange", type => {
+        if (type === "closed") {
+          serverResolve();
+        }
+      });
+
+      transport
+        .startBulkSend({
+          actor: "root",
+          type: "file-stream",
+          length: reallyLong.length,
+        })
+        .then(write_data);
+    },
+
+    onTransportClosed() {
+      do_throw("Transport closed before we expected");
+    },
+  };
+
+  transport.ready();
+
+  return Promise.all([clientDeferred, serverDeferred]);
+};
+
+/** * Test Utils ***/
+
+function verify() {
+  const reallyLong = really_long();
+
+  const inputFile = getTestTempFile("bulk-input");
+  const outputFile = getTestTempFile("bulk-output");
+
+  Assert.equal(inputFile.fileSize, reallyLong.length);
+  Assert.equal(outputFile.fileSize, reallyLong.length);
+
+  // Ensure output file contents actually match
+  return new Promise(resolve => {
+    NetUtil.asyncFetch(
+      {
+        uri: NetUtil.newURI(getTestTempFile("bulk-output")),
+        loadUsingSystemPrincipal: true,
+      },
+      input => {
+        const outputData = NetUtil.readInputStreamToString(
+          input,
+          reallyLong.length
+        );
+        // Avoid do_check_eq here so we don't log the contents
+        Assert.ok(outputData === reallyLong);
+        input.close();
+        resolve();
+      }
+    );
+  }).then(cleanup_files);
+}
+
+function cleanup_files() {
+  const inputFile = getTestTempFile("bulk-input", true);
+  if (inputFile.exists()) {
+    inputFile.remove(false);
+  }
+
+  const outputFile = getTestTempFile("bulk-output", true);
+  if (outputFile.exists()) {
+    outputFile.remove(false);
+  }
+}
diff --git a/devtools/shared/transport/tests/xpcshell/testactors.js b/devtools/shared/transport/tests/xpcshell/testactors.js
new file mode 100644
index 0000000000..0a35f05287
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/testactors.js
@@ -0,0 +1,16 @@
+/* Any copyright is dedicated to the Public Domain.
+   http://creativecommons.org/publicdomain/zero/1.0/ */
+"use strict";
+
+const { RootActor } = require("resource://devtools/server/actors/root.js");
+const {
+  ActorRegistry,
+} = require("resource://devtools/server/actors/utils/actor-registry.js");
+
+exports.createRootActor = function createRootActor(connection) {
+  const root = new RootActor(connection, {
+    globalActorFactories: ActorRegistry.globalActorFactories,
+  });
+  root.applicationType = "xpcshell-tests";
+  return root;
+};
diff --git a/devtools/shared/transport/tests/xpcshell/xpcshell.ini b/devtools/shared/transport/tests/xpcshell/xpcshell.ini
new file mode 100644
index 0000000000..fc369615c2
--- /dev/null
+++ b/devtools/shared/transport/tests/xpcshell/xpcshell.ini
@@ -0,0 +1,17 @@
+[DEFAULT]
+tags = devtools
+head = head_dbg.js
+firefox-appdir = browser
+skip-if = toolkit == 'android'
+
+support-files =
+  testactors.js
+
+[test_bulk_error.js]
+[test_client_server_bulk.js]
+[test_dbgsocket.js]
+[test_dbgsocket_connection_drop.js]
+[test_delimited_read.js]
+[test_packet.js]
+[test_queue.js]
+[test_transport_bulk.js]
diff --git a/devtools/shared/transport/transport.js b/devtools/shared/transport/transport.js
new file mode 100644
index 0000000000..460f45206f
--- /dev/null
+++ b/devtools/shared/transport/transport.js
@@ -0,0 +1,499 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+const DevToolsUtils = require("resource://devtools/shared/DevToolsUtils.js");
+const { dumpn, dumpv } = DevToolsUtils;
+const flags = require("resource://devtools/shared/flags.js");
+const StreamUtils = require("resource://devtools/shared/transport/stream-utils.js");
+const {
+  Packet,
+  JSONPacket,
+  BulkPacket,
+} = require("resource://devtools/shared/transport/packets.js");
+
+loader.lazyGetter(this, "ScriptableInputStream", () => {
+  return Components.Constructor(
+    "@mozilla.org/scriptableinputstream;1",
+    "nsIScriptableInputStream",
+    "init"
+  );
+});
+
+const PACKET_HEADER_MAX = 200;
+
+/**
+ * An adapter that handles data transfers between the devtools client and
+ * server. It can work with both nsIPipe and nsIServerSocket transports so
+ * long as the properly created input and output streams are specified.
+ * (However, for intra-process connections, LocalDebuggerTransport, below,
+ * is more efficient than using an nsIPipe pair with DebuggerTransport.)
+ *
+ * @param input nsIAsyncInputStream
+ *        The input stream.
+ * @param output nsIAsyncOutputStream
+ *        The output stream.
+ *
+ * Given a DebuggerTransport instance dt:
+ * 1) Set dt.hooks to a packet handler object (described below).
+ * 2) Call dt.ready() to begin watching for input packets.
+ * 3) Call dt.send() / dt.startBulkSend() to send packets.
+ * 4) Call dt.close() to close the connection, and disengage from the event
+ *    loop.
+ *
+ * A packet handler is an object with the following methods:
+ *
+ * - onPacket(packet) - called when we have received a complete packet.
+ *   |packet| is the parsed form of the packet --- a JavaScript value, not
+ *   a JSON-syntax string.
+ *
+ * - onBulkPacket(packet) - called when we have switched to bulk packet
+ *   receiving mode. |packet| is an object containing:
+ *   * actor:  Name of actor that will receive the packet
+ *   * type:   Name of actor's method that should be called on receipt
+ *   * length: Size of the data to be read
+ *   * stream: This input stream should only be used directly if you can ensure
+ *             that you will read exactly |length| bytes and will not close the
+ *             stream when reading is complete
+ *   * done:   If you use the stream directly (instead of |copyTo| below), you
+ *             must signal completion by resolving / rejecting this deferred.
+ *             If it's rejected, the transport will be closed.  If an Error is
+ *             supplied as a rejection value, it will be logged via |dumpn|.
+ *             If you do use |copyTo|, resolving is taken care of for you when
+ *             copying completes.
+ *   * copyTo: A helper function for getting your data out of the stream that
+ *             meets the stream handling requirements above, and has the
+ *             following signature:
+ *     @param  output nsIAsyncOutputStream
+ *             The stream to copy to.
+ *     @return Promise
+ *             The promise is resolved when copying completes or rejected if any
+ *             (unexpected) errors occur.
+ *             This object also emits "progress" events for each chunk that is
+ *             copied.  See stream-utils.js.
+ *
+ * - onTransportClosed(reason) - called when the connection is closed. |reason| is
+ *   an optional nsresult or object, typically passed when the transport is
+ *   closed due to some error in a underlying stream.
+ *
+ * See ./packets.js and the Remote Debugging Protocol specification for more
+ * details on the format of these packets.
+ */
+function DebuggerTransport(input, output) {
+  this._input = input;
+  this._scriptableInput = new ScriptableInputStream(input);
+  this._output = output;
+
+  // The current incoming (possibly partial) header, which will determine which
+  // type of Packet |_incoming| below will become.
+  this._incomingHeader = "";
+  // The current incoming Packet object
+  this._incoming = null;
+  // A queue of outgoing Packet objects
+  this._outgoing = [];
+
+  this.hooks = null;
+  this.active = false;
+
+  this._incomingEnabled = true;
+  this._outgoingEnabled = true;
+
+  this.close = this.close.bind(this);
+}
+
+DebuggerTransport.prototype = {
+  /**
+   * Transmit an object as a JSON packet.
+   *
+   * This method returns immediately, without waiting for the entire
+   * packet to be transmitted, registering event handlers as needed to
+   * transmit the entire packet. Packets are transmitted in the order
+   * they are passed to this method.
+   */
+  send(object) {
+    const packet = new JSONPacket(this);
+    packet.object = object;
+    this._outgoing.push(packet);
+    this._flushOutgoing();
+  },
+
+  /**
+   * Transmit streaming data via a bulk packet.
+   *
+   * This method initiates the bulk send process by queuing up the header data.
+   * The caller receives eventual access to a stream for writing.
+   *
+   * N.B.: Do *not* attempt to close the stream handed to you, as it will
+   * continue to be used by this transport afterwards.  Most users should
+   * instead use the provided |copyFrom| function instead.
+   *
+   * @param header Object
+   *        This is modeled after the format of JSON packets above, but does not
+   *        actually contain the data, but is instead just a routing header:
+   *          * actor:  Name of actor that will receive the packet
+   *          * type:   Name of actor's method that should be called on receipt
+   *          * length: Size of the data to be sent
+   * @return Promise
+   *         The promise will be resolved when you are allowed to write to the
+   *         stream with an object containing:
+   *           * stream:   This output stream should only be used directly if
+   *                       you can ensure that you will write exactly |length|
+   *                       bytes and will not close the stream when writing is
+   *                       complete
+   *           * done:     If you use the stream directly (instead of |copyFrom|
+   *                       below), you must signal completion by resolving /
+   *                       rejecting this deferred.  If it's rejected, the
+   *                       transport will be closed.  If an Error is supplied as
+   *                       a rejection value, it will be logged via |dumpn|.  If
+   *                       you do use |copyFrom|, resolving is taken care of for
+   *                       you when copying completes.
+   *           * copyFrom: A helper function for getting your data onto the
+   *                       stream that meets the stream handling requirements
+   *                       above, and has the following signature:
+   *             @param  input nsIAsyncInputStream
+   *                     The stream to copy from.
+   *             @return Promise
+   *                     The promise is resolved when copying completes or
+   *                     rejected if any (unexpected) errors occur.
+   *                     This object also emits "progress" events for each chunk
+   *                     that is copied.  See stream-utils.js.
+   */
+  startBulkSend(header) {
+    const packet = new BulkPacket(this);
+    packet.header = header;
+    this._outgoing.push(packet);
+    this._flushOutgoing();
+    return packet.streamReadyForWriting;
+  },
+
+  /**
+   * Close the transport.
+   * @param reason nsresult / object (optional)
+   *        The status code or error message that corresponds to the reason for
+   *        closing the transport (likely because a stream closed or failed).
+   */
+  close(reason) {
+    this.active = false;
+    this._input.close();
+    this._scriptableInput.close();
+    this._output.close();
+    this._destroyIncoming();
+    this._destroyAllOutgoing();
+    if (this.hooks) {
+      this.hooks.onTransportClosed(reason);
+      this.hooks = null;
+    }
+    if (reason) {
+      dumpn("Transport closed: " + DevToolsUtils.safeErrorString(reason));
+    } else {
+      dumpn("Transport closed.");
+    }
+  },
+
+  /**
+   * The currently outgoing packet (at the top of the queue).
+   */
+  get _currentOutgoing() {
+    return this._outgoing[0];
+  },
+
+  /**
+   * Flush data to the outgoing stream.  Waits until the output stream notifies
+   * us that it is ready to be written to (via onOutputStreamReady).
+   */
+  _flushOutgoing() {
+    if (!this._outgoingEnabled || this._outgoing.length === 0) {
+      return;
+    }
+
+    // If the top of the packet queue has nothing more to send, remove it.
+    if (this._currentOutgoing.done) {
+      this._finishCurrentOutgoing();
+    }
+
+    if (this._outgoing.length) {
+      const threadManager = Cc["@mozilla.org/thread-manager;1"].getService();
+      this._output.asyncWait(this, 0, 0, threadManager.currentThread);
+    }
+  },
+
+  /**
+   * Pause this transport's attempts to write to the output stream.  This is
+   * used when we've temporarily handed off our output stream for writing bulk
+   * data.
+   */
+  pauseOutgoing() {
+    this._outgoingEnabled = false;
+  },
+
+  /**
+   * Resume this transport's attempts to write to the output stream.
+   */
+  resumeOutgoing() {
+    this._outgoingEnabled = true;
+    this._flushOutgoing();
+  },
+
+  // nsIOutputStreamCallback
+  /**
+   * This is called when the output stream is ready for more data to be written.
+   * The current outgoing packet will attempt to write some amount of data, but
+   * may not complete.
+   */
+  onOutputStreamReady: DevToolsUtils.makeInfallible(function (stream) {
+    if (!this._outgoingEnabled || this._outgoing.length === 0) {
+      return;
+    }
+
+    try {
+      this._currentOutgoing.write(stream);
+    } catch (e) {
+      if (e.result != Cr.NS_BASE_STREAM_WOULD_BLOCK) {
+        this.close(e.result);
+        return;
+      }
+      throw e;
+    }
+
+    this._flushOutgoing();
+  }, "DebuggerTransport.prototype.onOutputStreamReady"),
+
+  /**
+   * Remove the current outgoing packet from the queue upon completion.
+   */
+  _finishCurrentOutgoing() {
+    if (this._currentOutgoing) {
+      this._currentOutgoing.destroy();
+      this._outgoing.shift();
+    }
+  },
+
+  /**
+   * Clear the entire outgoing queue.
+   */
+  _destroyAllOutgoing() {
+    for (const packet of this._outgoing) {
+      packet.destroy();
+    }
+    this._outgoing = [];
+  },
+
+  /**
+   * Initialize the input stream for reading. Once this method has been called,
+   * we watch for packets on the input stream, and pass them to the appropriate
+   * handlers via this.hooks.
+   */
+  ready() {
+    this.active = true;
+    this._waitForIncoming();
+  },
+
+  /**
+   * Asks the input stream to notify us (via onInputStreamReady) when it is
+   * ready for reading.
+   */
+  _waitForIncoming() {
+    if (this._incomingEnabled) {
+      const threadManager = Cc["@mozilla.org/thread-manager;1"].getService();
+      this._input.asyncWait(this, 0, 0, threadManager.currentThread);
+    }
+  },
+
+  /**
+   * Pause this transport's attempts to read from the input stream.  This is
+   * used when we've temporarily handed off our input stream for reading bulk
+   * data.
+   */
+  pauseIncoming() {
+    this._incomingEnabled = false;
+  },
+
+  /**
+   * Resume this transport's attempts to read from the input stream.
+   */
+  resumeIncoming() {
+    this._incomingEnabled = true;
+    this._flushIncoming();
+    this._waitForIncoming();
+  },
+
+  // nsIInputStreamCallback
+  /**
+   * Called when the stream is either readable or closed.
+   */
+  onInputStreamReady: DevToolsUtils.makeInfallible(function (stream) {
+    try {
+      while (
+        stream.available() &&
+        this._incomingEnabled &&
+        this._processIncoming(stream, stream.available())
+      ) {
+        // Loop until there is nothing more to process
+      }
+      this._waitForIncoming();
+    } catch (e) {
+      if (e.result != Cr.NS_BASE_STREAM_WOULD_BLOCK) {
+        this.close(e.result);
+      } else {
+        throw e;
+      }
+    }
+  }, "DebuggerTransport.prototype.onInputStreamReady"),
+
+  /**
+   * Process the incoming data.  Will create a new currently incoming Packet if
+   * needed.  Tells the incoming Packet to read as much data as it can, but
+   * reading may not complete.  The Packet signals that its data is ready for
+   * delivery by calling one of this transport's _on*Ready methods (see
+   * ./packets.js and the _on*Ready methods below).
+   * @return boolean
+   *         Whether incoming stream processing should continue for any
+   *         remaining data.
+   */
+  _processIncoming(stream, count) {
+    dumpv("Data available: " + count);
+
+    if (!count) {
+      dumpv("Nothing to read, skipping");
+      return false;
+    }
+
+    try {
+      if (!this._incoming) {
+        dumpv("Creating a new packet from incoming");
+
+        if (!this._readHeader(stream)) {
+          // Not enough data to read packet type
+          return false;
+        }
+
+        // Attempt to create a new Packet by trying to parse each possible
+        // header pattern.
+        this._incoming = Packet.fromHeader(this._incomingHeader, this);
+        if (!this._incoming) {
+          throw new Error(
+            "No packet types for header: " + this._incomingHeader
+          );
+        }
+      }
+
+      if (!this._incoming.done) {
+        // We have an incomplete packet, keep reading it.
+        dumpv("Existing packet incomplete, keep reading");
+        this._incoming.read(stream, this._scriptableInput);
+      }
+    } catch (e) {
+      const msg =
+        "Error reading incoming packet: (" + e + " - " + e.stack + ")";
+      dumpn(msg);
+
+      // Now in an invalid state, shut down the transport.
+      this.close();
+      return false;
+    }
+
+    if (!this._incoming.done) {
+      // Still not complete, we'll wait for more data.
+      dumpv("Packet not done, wait for more");
+      return true;
+    }
+
+    // Ready for next packet
+    this._flushIncoming();
+    return true;
+  },
+
+  /**
+   * Read as far as we can into the incoming data, attempting to build up a
+   * complete packet header (which terminates with ":").  We'll only read up to
+   * PACKET_HEADER_MAX characters.
+   * @return boolean
+   *         True if we now have a complete header.
+   */
+  _readHeader() {
+    const amountToRead = PACKET_HEADER_MAX - this._incomingHeader.length;
+    this._incomingHeader += StreamUtils.delimitedRead(
+      this._scriptableInput,
+      ":",
+      amountToRead
+    );
+    if (flags.wantVerbose) {
+      dumpv("Header read: " + this._incomingHeader);
+    }
+
+    if (this._incomingHeader.endsWith(":")) {
+      if (flags.wantVerbose) {
+        dumpv("Found packet header successfully: " + this._incomingHeader);
+      }
+      return true;
+    }
+
+    if (this._incomingHeader.length >= PACKET_HEADER_MAX) {
+      throw new Error("Failed to parse packet header!");
+    }
+
+    // Not enough data yet.
+    return false;
+  },
+
+  /**
+   * If the incoming packet is done, log it as needed and clear the buffer.
+   */
+  _flushIncoming() {
+    if (!this._incoming.done) {
+      return;
+    }
+    if (flags.wantLogging) {
+      dumpn("Got: " + this._incoming);
+    }
+    this._destroyIncoming();
+  },
+
+  /**
+   * Handler triggered by an incoming JSONPacket completing it's |read| method.
+   * Delivers the packet to this.hooks.onPacket.
+   */
+  _onJSONObjectReady(object) {
+    DevToolsUtils.executeSoon(
+      DevToolsUtils.makeInfallible(() => {
+        // Ensure the transport is still alive by the time this runs.
+        if (this.active) {
+          this.hooks.onPacket(object);
+        }
+      }, "DebuggerTransport instance's this.hooks.onPacket")
+    );
+  },
+
+  /**
+   * Handler triggered by an incoming BulkPacket entering the |read| phase for
+   * the stream portion of the packet.  Delivers info about the incoming
+   * streaming data to this.hooks.onBulkPacket.  See the main comment on the
+   * transport at the top of this file for more details.
+   */
+  _onBulkReadReady(...args) {
+    DevToolsUtils.executeSoon(
+      DevToolsUtils.makeInfallible(() => {
+        // Ensure the transport is still alive by the time this runs.
+        if (this.active) {
+          this.hooks.onBulkPacket(...args);
+        }
+      }, "DebuggerTransport instance's this.hooks.onBulkPacket")
+    );
+  },
+
+  /**
+   * Remove all handlers and references related to the current incoming packet,
+   * either because it is now complete or because the transport is closing.
+   */
+  _destroyIncoming() {
+    if (this._incoming) {
+      this._incoming.destroy();
+    }
+    this._incomingHeader = "";
+    this._incoming = null;
+  },
+};
+
+exports.DebuggerTransport = DebuggerTransport;
diff --git a/devtools/shared/transport/websocket-transport.js b/devtools/shared/transport/websocket-transport.js
new file mode 100644
index 0000000000..b8255e0067
--- /dev/null
+++ b/devtools/shared/transport/websocket-transport.js
@@ -0,0 +1,86 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+const EventEmitter = require("resource://devtools/shared/event-emitter.js");
+
+function WebSocketDebuggerTransport(socket) {
+  EventEmitter.decorate(this);
+
+  this.active = false;
+  this.hooks = null;
+  this.socket = socket;
+}
+
+WebSocketDebuggerTransport.prototype = {
+  ready() {
+    if (this.active) {
+      return;
+    }
+
+    this.socket.addEventListener("message", this);
+    this.socket.addEventListener("close", this);
+
+    this.active = true;
+  },
+
+  send(object) {
+    this.emit("send", object);
+    if (this.socket) {
+      this.socket.send(JSON.stringify(object));
+    }
+  },
+
+  startBulkSend() {
+    throw new Error("Bulk send is not supported by WebSocket transport");
+  },
+
+  close() {
+    if (!this.socket) {
+      return;
+    }
+    this.emit("close");
+    this.active = false;
+
+    this.socket.removeEventListener("message", this);
+    this.socket.removeEventListener("close", this);
+    this.socket.close();
+    this.socket = null;
+
+    if (this.hooks) {
+      if (this.hooks.onTransportClosed) {
+        this.hooks.onTransportClosed();
+      }
+      this.hooks = null;
+    }
+  },
+
+  handleEvent(event) {
+    switch (event.type) {
+      case "message":
+        this.onMessage(event);
+        break;
+      case "close":
+        this.close();
+        break;
+    }
+  },
+
+  onMessage({ data }) {
+    if (typeof data !== "string") {
+      throw new Error(
+        "Binary messages are not supported by WebSocket transport"
+      );
+    }
+
+    const object = JSON.parse(data);
+    this.emit("packet", object);
+    if (this.hooks) {
+      this.hooks.onPacket(object);
+    }
+  },
+};
+
+module.exports = WebSocketDebuggerTransport;
diff --git a/devtools/shared/transport/worker-transport.js b/devtools/shared/transport/worker-transport.js
new file mode 100644
index 0000000000..903fd69cf4
--- /dev/null
+++ b/devtools/shared/transport/worker-transport.js
@@ -0,0 +1,113 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+"use strict";
+
+// Each worker debugger supports only a single connection to the main thread.
+// However, its theoretically possible for multiple servers to connect to the
+// same worker. Consequently, each transport has a connection id, to allow
+// messages from multiple connections to be multiplexed on a single channel.
+
+/**
+ * A transport that uses a WorkerDebugger to send packets from the main
+ * thread to a worker thread.
+ */
+class MainThreadWorkerDebuggerTransport {
+  constructor(dbg, id) {
+    this._dbg = dbg;
+    this._id = id;
+
+    this._dbgListener = {
+      onMessage: this._onMessage.bind(this),
+    };
+  }
+
+  ready() {
+    this._dbg.addListener(this._dbgListener);
+  }
+
+  close() {
+    if (this._dbgListener) {
+      this._dbg.removeListener(this._dbgListener);
+    }
+    this._dbgListener = null;
+    this.hooks?.onTransportClosed();
+  }
+
+  send(packet) {
+    this._dbg.postMessage(
+      JSON.stringify({
+        type: "message",
+        id: this._id,
+        message: packet,
+      })
+    );
+  }
+
+  startBulkSend() {
+    throw new Error("Can't send bulk data from worker threads!");
+  }
+
+  _onMessage(message) {
+    const packet = JSON.parse(message);
+    if (packet.type !== "message" || packet.id !== this._id || !this.hooks) {
+      return;
+    }
+
+    this.hooks.onPacket(packet.message);
+  }
+}
+
+exports.MainThreadWorkerDebuggerTransport = MainThreadWorkerDebuggerTransport;
+
+/**
+ * A transport that uses a WorkerDebuggerGlobalScope to send packets from a
+ * worker thread to the main thread.
+ */
+function WorkerThreadWorkerDebuggerTransport(scope, id) {
+  this._scope = scope;
+  this._id = id;
+  this._onMessage = this._onMessage.bind(this);
+}
+
+WorkerThreadWorkerDebuggerTransport.prototype = {
+  constructor: WorkerThreadWorkerDebuggerTransport,
+
+  ready() {
+    this._scope.addEventListener("message", this._onMessage);
+  },
+
+  close() {
+    this._scope.removeEventListener("message", this._onMessage);
+    this.hooks?.onTransportClosed();
+  },
+
+  send(packet) {
+    this._scope.postMessage(
+      JSON.stringify({
+        type: "message",
+        id: this._id,
+        message: packet,
+      })
+    );
+  },
+
+  startBulkSend() {
+    throw new Error("Can't send bulk data from worker threads!");
+  },
+
+  _onMessage(event) {
+    const packet = JSON.parse(event.data);
+    if (packet.type !== "message" || packet.id !== this._id) {
+      return;
+    }
+
+    if (this.hooks) {
+      this.hooks.onPacket(packet.message);
+    }
+  },
+};
+
+exports.WorkerThreadWorkerDebuggerTransport =
+  WorkerThreadWorkerDebuggerTransport;