Abort handling

karma.global.js

@@ -36,7 +36,6 @@ module.exports = (config) => ({
       format: 'iife', // Helps prevent naming collisions.
       name: 'confluxTest', // Required for 'iife' format.
       sourcemap: 'inline', // Sensible for testing.
-      file: 'dist/test2.js',
     },
     plugins: [
       resolve({

test/write.test.js

@@ -1,13 +1,77 @@
 import test from 'tape';
 import { Writer, Reader } from '../src/index.js';
 
-function streamFrom(chunks) {
-  return new ReadableStream({
+// Constants
+const ABORT_MESSAGE = 'This is a bad stream';
+
+function writeStuff(writable, inputReadable) {
+  const writer = writable.getWriter();
+  writer.write({
+    name: "/chonks.txt",
+    stream: () => inputReadable,
+  });
+  writer.close();
+  return writer;
+}
+
+function getInputReadable({
+  chunks = [[1, 2, 3, 4], [5, 6, 7, 8]],
+  abort = false,
+  long = false,
+} = {}) {
+  let cancelReason;
+
+  const inputReadable = new ReadableStream({
+    start(ctrl) {
+      if (long) {
+        for (let x = 0; x < 1000; x++) ctrl.enqueue(chunks[0]);
+      }
+
+      if (abort) {
+        ctrl.error(ABORT_MESSAGE);
+      } else {
+        chunks.forEach((chunk) => ctrl.enqueue(chunk));
+        ctrl.close();
+      }
+    },
+
+    cancel(reason) {
+      cancelReason = reason;
+    }
+  });
+
+  inputReadable.cancelReason = cancelReason;
+  return inputReadable;
+}
+
+function getOutputWritable({ abort = false } = {}) {
+  let chunks = [];
+  let abortReason;
+  let closed;
+
+  const outputWritable = new WritableStream({
     start(ctrl) {
-      chunks.forEach((chunk) => ctrl.enqueue(chunk));
-      ctrl.close();
+      closed = false;
+      if (abort) ctrl.error(ABORT_MESSAGE);
+    },
+
+    write(chunk) {
+      chunks.push(chunk);
     },
+
+    abort(reason) {
+      abortReason = reason;
+    },
+
+    close() {
+      closed = true
+    }
   });
+
+  outputWritable.outputChunks = chunks;
+  outputWritable.abortReason = abortReason;
+  outputWritable.closed = closed;
+  return outputWritable;
 }
 
 const date = +new Date('2012-02-05T15:40:48Z');
@@ -95,3 +159,126 @@ test('Writing - All in one big test', async (t) => {
 
   t.end();
 });
+
+test('Writing - from custom stream', async (t) => {
+  const { readable, writable } = new Writer();
+  const inputChunks = [[1, 2, 3, 4], [5, 6, 7, 8]];
+
+  const inputReadable = getInputReadable({ inputChunks })
+  const writer = writeStuff(writable, inputReadable);
+
+  const outputWritable = getOutputWritable();
+  await readable.pipeTo(outputWritable);
+
+  // Extract stored chunks off custom property on stream object
+  let { outputChunks } = outputWritable;
+
+  // Remove header and footer
+  outputChunks = outputChunks.slice(1, inputChunks.length + 1);
+
+  t.deepEqual(outputChunks, inputChunks);
+});
+
+test('Writing - abort input readable stream', async (t) => {
+  const { readable, writable } = new Writer();
+
+  const inputReadable = getInputReadable({ abort: true });
+  const writer = writeStuff(writable, inputReadable);
+
+  const outputWritable = getOutputWritable();
+
+  try {
+    await readable.pipeTo(outputWritable);
+  } catch(error) {
+    t.equals(error, ABORT_MESSAGE);
+  }
+});
+
+test('Writing - cancel input readable stream', async (t) => {
+  const { readable, writable } = new Writer();
+
+  const inputReadable = getInputReadable();
+  inputReadable.cancel(ABORT_MESSAGE);
+  const writer = writeStuff(writable, inputReadable);
+
+  const outputWritable = getOutputWritable();
+
+  try {
+    await readable.pipeTo(outputWritable);
+  } catch(error) {
+    const { cancelReason } = inputReadable;
+    const { abortReason } = outputWritable;
+
+    t.equals(error, ABORT_MESSAGE);
+    t.equals(abortReason, ABORT_MESSAGE);
+    t.equals(cancelReason, ABORT_MESSAGE);
+  }
+});
+
+test('Writing - abort output writable stream', async (t) => {
+  const { readable, writable } = new Writer();
+
+  const inputReadable = getInputReadable();
+  const writer = writeStuff(writable, inputReadable);
+
+  const outputWritable = getOutputWritable({ long: true });
+
+  try {
+    const promise = readable.pipeTo(outputWritable);
+    outputWritable.abort(ABORT_MESSAGE);
+    await promise;
+  } catch(error) {
+    const { cancelReason } = inputReadable;
+    const { abortReason } = outputWritable;
+
+    t.equals(error, ABORT_MESSAGE);
+    t.equals(abortReason, ABORT_MESSAGE);
+    t.equals(cancelReason, ABORT_MESSAGE);
+  }
+});
+
+test('Writing - early close output writable stream', async (t) => {
+  const { readable, writable } = new Writer();
+
+  const inputReadable = getInputReadable();
+  const writer = writeStuff(writable, inputReadable);
+
+  const outputWritable = getOutputWritable({ long: true });
+
+  try {
+    const promise = readable.pipeTo(outputWritable);
+    outputWritable.close();
+    await promise;
+  } catch(error) {
+    const { cancelReason } = inputReadable;
+    const { abortReason } = outputWritable;
+
+    t.equals(error, ABORT_MESSAGE);
+    t.equals(abortReason, ABORT_MESSAGE);
+    t.equals(cancelReason, ABORT_MESSAGE);
+  }
+});
+
+test('Writing - error inoutput writable stream', async (t) => {
+  const { readable, writable } = new Writer();
+
+  const inputReadable = getInputReadable();
+  const writer = writeStuff(writable, inputReadable);
+
+  const outputWritable = getOutputWritable({ abort: true });
+
+  try {
+    const promise = readable.pipeTo(outputWritable);
+    await promise;
+  } catch(error) {
+    const { cancelReason } = inputReadable;
+    const { abortReason } = outputWritable;
+
+    t.equals(error, ABORT_MESSAGE);
+    // t.equals(abortReason, ABORT_MESSAGE);
+    // t.equals(cancelReason, ABORT_MESSAGE);
+  }
+});
+
+
+