firefox/dom/tests/mochitest/fetch/common_readableStreams.js

const SAME_COMPARTMENT = "same-compartment";
const IFRAME_COMPARTMENT = "iframe-compartment";
const BIG_BUFFER_SIZE = 1000000;
const ITER_MAX = 10;

function makeBuffer(size) {
  let buffer = new Uint8Array(size);
  buffer.fill(42);

  let value = 0;
  for (let i = 0; i < 1000000; i += 1000) {
    buffer.set([++value % 255], i);
  }

  return buffer;
}

function apply_compartment(compartment, data) {
  if (compartment == SAME_COMPARTMENT) {
    return self[data.func](data.args, self);
  }

  if (compartment == IFRAME_COMPARTMENT) {
    const iframe = document.querySelector("#iframe").contentWindow;
    return iframe[data.func](data.args, self);
  }

  ok(false, "Invalid compartment value");
  return undefined;
}

async function test_nativeStream(compartment) {
  info("test_nativeStream");

  let r = await fetch("/");

  return apply_compartment(compartment, {
    func: "test_nativeStream_continue",
    args: r,
  });
}

async function test_nativeStream_continue(r, that) {
  that.ok(r.body instanceof that.ReadableStream, "We have a ReadableStream");

  let a = r.clone();
  that.ok(a instanceof that.Response, "We have a cloned Response");
  that.ok(a.body instanceof that.ReadableStream, "We have a ReadableStream");

  let b = a.clone();
  that.ok(b instanceof that.Response, "We have a cloned Response");
  that.ok(b.body instanceof that.ReadableStream, "We have a ReadableStream");

  let blob = await r.blob();

  that.ok(blob instanceof that.Blob, "We have a blob");
  let d = await a.body.getReader().read();

  that.ok(!d.done, "We have read something!");
  blob = await b.blob();

  that.ok(blob instanceof that.Blob, "We have a blob");
}

async function test_timeout(compartment) {
  info("test_timeout");

  let blob = new Blob([""]);
  let r = await fetch(URL.createObjectURL(blob));

  return apply_compartment(compartment, {
    func: "test_timeout_continue",
    args: r,
  });
}

async function test_timeout_continue(r, that) {
  await r.body.getReader().read();

  await new Promise(resolve => setTimeout(resolve, 0));

  try {
    await r.blob();
    that.ok(false, "We cannot have a blob here!");
  } catch (exc) {
    that.ok(true, "We cannot have a blob here!");
  }
}

async function test_nonNativeStream(compartment) {
  info("test_nonNativeStream");

  let buffer = makeBuffer(BIG_BUFFER_SIZE);
  info("Buffer size: " + buffer.byteLength);

  let r = new Response(
    new ReadableStream({
      start: controller => {
        controller.enqueue(buffer);
        controller.close();
      },
    })
  );

  return apply_compartment(compartment, {
    func: "test_nonNativeStream_continue",
    args: { r, buffer },
  });
}

async function test_nonNativeStream_continue(data, that) {
  that.ok(
    data.r.body instanceof that.ReadableStream,
    "We have a ReadableStream"
  );

  let a = data.r.clone();
  that.ok(a instanceof that.Response, "We have a cloned Response");
  that.ok(a.body instanceof that.ReadableStream, "We have a ReadableStream");

  let b = a.clone();
  that.ok(b instanceof that.Response, "We have a cloned Response");
  that.ok(b.body instanceof that.ReadableStream, "We have a ReadableStream");

  let blob = await data.r.blob();

  that.ok(blob instanceof that.Blob, "We have a blob");
  let d = await a.body.getReader().read();

  that.ok(!d.done, "We have read something!");
  blob = await b.blob();

  that.ok(blob instanceof that.Blob, "We have a blob");
  that.is(blob.size, data.buffer.byteLength, "Blob size matches");
}

async function test_noUint8Array(compartment) {
  info("test_noUint8Array");

  let r = new Response(
    new ReadableStream({
      start: controller => {
        controller.enqueue("hello world!");
        controller.close();
      },
    })
  );

  return apply_compartment(compartment, {
    func: "test_noUint8Array_continue",
    args: r,
  });
}

async function test_noUint8Array_continue(r, that) {
  that.ok(r.body instanceof that.ReadableStream, "We have a ReadableStream");

  try {
    await r.blob();
    that.ok(false, "We cannot have a blob here!");
  } catch (ex) {
    that.ok(true, "We cannot have a blob here!");
  }
}

async function test_pendingStream(compartment) {
  let r = new Response(
    new ReadableStream({
      start: controller => {
        controller.enqueue(makeBuffer(BIG_BUFFER_SIZE));
        // Let's keep this controler open.
        self.ccc = controller;
      },
    })
  );

  return apply_compartment(compartment, {
    func: "test_pendingStream_continue",
    args: r,
  });
}

async function test_pendingStream_continue(r, that) {
  let d = await r.body.getReader().read();

  that.ok(!d.done, "We have read something!");

  if ("close" in that) {
    that.close();
  }
}

async function test_nativeStream_cache(compartment) {
  info("test_nativeStream_cache");

  let origBody = "123456789abcdef";
  let url = "/nativeStream";

  let cache = await caches.open("nativeStream");

  info("Storing a body as a string");
  await cache.put(url, new Response(origBody));

  return apply_compartment(compartment, {
    func: "test_nativeStream_cache_continue",
    args: { caches, cache, url, origBody },
  });
}

async function test_nativeStream_cache_continue(data, that) {
  that.info("Retrieving the stored value");
  let cacheResponse = await data.cache.match(data.url);

  that.info("Converting the response to text");
  let cacheBody = await cacheResponse.text();

  that.is(data.origBody, cacheBody, "Bodies match");

  await data.caches.delete("nativeStream");
}

async function test_nonNativeStream_cache(compartment) {
  info("test_nonNativeStream_cache");

  let url = "/nonNativeStream";

  let cache = await caches.open("nonNativeStream");
  let buffer = makeBuffer(BIG_BUFFER_SIZE);
  info("Buffer size: " + buffer.byteLength);

  info("Storing a body as a string");
  let r = new Response(
    new ReadableStream({
      start: controller => {
        controller.enqueue(buffer);
        controller.close();
      },
    })
  );

  return apply_compartment(compartment, {
    func: "test_nonNativeStream_cache_continue",
    args: { caches, cache, buffer, r },
  });
}

async function test_nonNativeStream_cache_continue(data, that) {
  await data.cache.put(data.url, data.r);

  that.info("Retrieving the stored value");
  let cacheResponse = await data.cache.match(data.url);

  that.info("Converting the response to text");
  let cacheBody = await cacheResponse.arrayBuffer();

  that.ok(cacheBody instanceof that.ArrayBuffer, "Body is an array buffer");
  that.is(cacheBody.byteLength, BIG_BUFFER_SIZE, "Body length is correct");

  let value = 0;
  for (let i = 0; i < 1000000; i += 1000) {
    that.is(
      new Uint8Array(cacheBody)[i],
      ++value % 255,
      "byte in position " + i + " is correct"
    );
  }

  await data.caches.delete("nonNativeStream");
}

async function test_codeExecution(compartment) {
  info("test_codeExecution");

  let r = new Response(
    new ReadableStream({
      start(c) {
        controller = c;
      },
      pull() {
        console.log("pull called");
      },
    })
  );

  return apply_compartment(compartment, {
    func: "test_codeExecution_continue",
    args: r,
  });
}

// This is intended to just be a drop-in replacement for an old observer
// notification.
function addConsoleStorageListener(listener) {
  const ConsoleAPIStorage = SpecialPowers.Cc[
    "@mozilla.org/consoleAPI-storage;1"
  ].getService(SpecialPowers.Ci.nsIConsoleAPIStorage);
  listener.__handler = (message, id) => {
    listener.observe(message, id);
  };
  ConsoleAPIStorage.addLogEventListener(
    listener.__handler,
    SpecialPowers.wrap(document).nodePrincipal
  );
}

function removeConsoleStorageListener(listener) {
  const ConsoleAPIStorage = SpecialPowers.Cc[
    "@mozilla.org/consoleAPI-storage;1"
  ].getService(SpecialPowers.Ci.nsIConsoleAPIStorage);
  ConsoleAPIStorage.removeLogEventListener(listener.__handler);
}

async function test_codeExecution_continue(r, that) {
  function consoleListener() {
    addConsoleStorageListener(this);
  }

  var promise = new Promise(resolve => {
    consoleListener.prototype = {
      observe(aSubject) {
        that.ok(true, "Something has been received");

        var obj = aSubject.wrappedJSObject;
        if (obj.arguments[0] && obj.arguments[0] === "pull called") {
          that.ok(true, "Message received!");
          removeConsoleStorageListener(this);
          resolve();
        }
      },
    };
  });

  var cl = new consoleListener();

  r.body.getReader().read();
  await promise;
}

async function test_global(compartment) {
  info("test_global: " + compartment);

  self.foo = 42;
  self.iter = ITER_MAX;

  let r = new Response(
    new ReadableStream({
      start(c) {
        self.controller = c;
      },
      pull() {
        if (!("iter" in self) || self.iter < 0 || self.iter > ITER_MAX) {
          throw "Something bad is happening here!";
        }

        let buffer = new Uint8Array(1);
        buffer.fill(self.foo);
        self.controller.enqueue(buffer);

        if (--self.iter == 0) {
          controller.close();
        }
      },
    })
  );

  return apply_compartment(compartment, {
    func: "test_global_continue",
    args: r,
  });
}

async function test_global_continue(r, that) {
  let a = await r.arrayBuffer();

  that.is(
    Object.getPrototypeOf(a),
    that.ArrayBuffer.prototype,
    "Body is an array buffer"
  );
  that.is(a.byteLength, ITER_MAX, "Body length is correct");

  for (let i = 0; i < ITER_MAX; ++i) {
    that.is(new Uint8Array(a)[i], 42, "Byte " + i + " is correct");
  }
}

function workify(func) {
  info("Workifying " + func);

  return new Promise((resolve, reject) => {
    let worker = new Worker("worker_readableStreams.js");
    worker.postMessage(func);
    worker.onmessage = function (e) {
      if (e.data.type == "done") {
        resolve();
        return;
      }

      if (e.data.type == "error") {
        reject(e.data.message);
        return;
      }

      if (e.data.type == "test") {
        ok(e.data.test, e.data.message);
        return;
      }

      if (e.data.type == "info") {
        info(e.data.message);
      }
    };
  });
}