Initial commit

Author: zebp

.github/workflows/ci.yml

@@ -0,0 +1,29 @@
+name: ci
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        deno-version: [1.9.2]
+
+    steps:
+      - name: Git Checkout Deno Module
+        uses: actions/checkout@v2
+      - name: Use Deno Version ${{ matrix.deno-version }}
+        uses: denolib/setup-deno@v2
+        with:
+          deno-version: ${{ matrix.deno-version }}
+      - name: Lint Deno Module
+        run: deno fmt --check
+      - name: Build Deno Module
+        run: deno run --reload mod.ts
+      - name: Test Deno Module
+        run: deno test --allow-none

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+  "deno.enable": true,
+  "editor.tabSize": 2,
+  "[typescript]": {
+    "editor.defaultFormatter": "denoland.vscode-deno"
+  }
+}

README.md

@@ -0,0 +1,24 @@
+# multipart-stream
+
+Create ReadableStreams from multipart forms without allocating the entire form
+on the heap.
+
+## Example
+
+```typescript
+import { streamFromMultipart } from "https://deno.land/x/multipart_stream/mod.ts";
+
+const [stream, boundary] = streamFromMultipart(async (multipartWriter) => {
+  const file = await Deno.open("test.bin");
+  await multipartWriter.writeFile("file", "test.bin", file);
+  file.close();
+});
+
+await fetch("http://example.com/upload", {
+  headers: {
+    "Content-Type": `multipart/form-data; boundary=${boundary}`,
+  },
+  body: stream,
+  method: "POST",
+});
+```

mod.ts

@@ -0,0 +1,88 @@
+import { Channel } from "https://deno.land/x/channo@0.1.1/mod.ts";
+import {
+  MultipartWriter,
+} from "https://deno.land/std@0.95.0/mime/multipart.ts";
+import {
+  readableStreamFromIterable,
+  readableStreamFromReader,
+  readerFromStreamReader,
+} from "https://deno.land/std@0.95.0/io/streams.ts";
+
+interface BytesMessage {
+  type: "bytes";
+  buffer: Uint8Array;
+}
+
+interface ErrorMessage {
+  type: "error";
+  error: unknown;
+}
+
+interface DoneMessage {
+  type: "done";
+}
+
+type Message =
+  | BytesMessage
+  | ErrorMessage
+  | DoneMessage;
+
+/**
+ * Creates a {@link ReadableStream} by serializing a user populated {@link MultipartWriter}.
+ * 
+ * @param writerFunction A function that receives a prepared {@link MultipartWriter} that the user
+ * can append fields or files to.
+ * @returns A tuple of {@link ReadableStream} and the multipart boundary.
+ */
+export function streamFromMultipart(
+  writerFunction: (
+    multipartWriter: MultipartWriter,
+  ) => Promise<void>,
+): [ReadableStream<Uint8Array>, string] {
+  const channel = new Channel<Message>();
+
+  // Creates a writer where all of the data is passed to our channel so it can be drained to a
+  // ReadableStream.
+  const multipartWriter = new MultipartWriter({
+    write(buffer: Uint8Array): Promise<number> {
+      channel.push({ type: "bytes", buffer });
+      return Promise.resolve(buffer.length);
+    },
+  });
+
+  // Passes the multipart writer to the caller so they can populate it.
+  writerFunction(multipartWriter)
+    .then(() => {
+      try {
+        multipartWriter.close();
+      } catch (_ignored) {
+        // We'll try to close the writer incase the user hasn't, if they have the close function
+        // will throw an error we'll just ignore.
+      }
+    })
+    .then(() => channel.push({ type: "done" }))
+    .catch((error) => channel.push({ type: "error", error }));
+
+  // A generator that yields values pushed to our multipart writer.
+  async function* generator(): AsyncGenerator<Uint8Array, void, undefined> {
+    for await (const message of channel.stream()) {
+      if (message.type === "done") {
+        channel.close();
+        return;
+      } else if (message.type === "error") {
+        throw message.error;
+      }
+
+      yield message.buffer;
+    }
+  }
+
+  // Yes I know this looks REALLY stupid, but I was having issues where if we used the potentially
+  // broken stream we would send the data out of order. This fixes it but I have no idea why. It
+  // doesn't allocate the entire stream on the heap, so I think this is going to stay for now.
+  const potentiallyBrokenStream = readableStreamFromIterable(generator());
+  const reader = readerFromStreamReader(potentiallyBrokenStream.getReader());
+  const stream = readableStreamFromReader(reader);
+
+  return [stream, multipartWriter.boundary];
+}

test.ts

@@ -0,0 +1,37 @@
+import {
+  isFormFile,
+  MultipartReader,
+} from "https://deno.land/std@0.95.0/mime/multipart.ts";
+import {
+  assert,
+  assertEquals,
+} from "https://deno.land/std@0.95.0/testing/asserts.ts";
+import { Buffer } from "https://deno.land/std@0.95.0/io/buffer.ts";
+import { readerFromStreamReader } from "https://deno.land/std@0.95.0/io/streams.ts";
+import { streamFromMultipart } from "./mod.ts";
+
+const textEncoder = new TextEncoder();
+const textBytes = textEncoder.encode("denoland".repeat(1024));
+const textBytesReader = new Buffer(textBytes) as Deno.Reader;
+
+Deno.test({
+  name: "parse",
+  fn: async () => {
+    const [stream, boundary] = streamFromMultipart(async (writer) => {
+      await writer.writeFile("test", "test.bin", textBytesReader);
+      await writer.writeField("deno", "land");
+    });
+
+    const reader = readerFromStreamReader(stream.getReader());
+    const multipartReader = new MultipartReader(reader, boundary);
+    const form = await multipartReader.readForm();
+
+    // Ensure the file was serialized correctly.
+    const formFile = form.file("test");
+    assert(isFormFile(formFile), "form file is invalid");
+    assertEquals(formFile.content, textBytes);
+
+    // Ensure the field was serialized correctly.
+    assertEquals(form.value("deno"), "land");
+  },
+});