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); } }; }); }