Fix issue with potential stream corruption

Author: zebp

mod.ts

@@ -4,8 +4,6 @@ import {
 } 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 {
@@ -45,7 +43,9 @@ export function streamFromMultipart(
   // ReadableStream.
   const multipartWriter = new MultipartWriter({
     write(buffer: Uint8Array): Promise<number> {
-      channel.push({ type: "bytes", buffer });
+      // It is VERY important we close the buffer, there is no guarentee that the writer won't
+      // re-use this buffer for subsequent writes.
+      channel.push({ type: "bytes", buffer: new Uint8Array(buffer) });
       return Promise.resolve(buffer.length);
     },
   });
@@ -54,7 +54,8 @@ export function streamFromMultipart(
   writerFunction(multipartWriter)
     .then(() => {
       try {
-        multipartWriter.close();
+        // Close is an async function that still writes, so we must wait for it to complete.
+        return 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.
@@ -77,12 +78,5 @@ export function streamFromMultipart(
     }
   }
 
-  // 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];
+  return [readableStreamFromIterable(generator()), multipartWriter.boundary];
 }

test.ts

@@ -7,29 +7,29 @@ import {
   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 { readerFromIterable } 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;
+const testBytes = new Uint8Array(2 << 16).map(() =>
+  Math.round(Math.random() * 255)
+);
+const testBytesReader = new Buffer(testBytes) as Deno.Reader;
 
 Deno.test({
   name: "parse",
   fn: async () => {
     const [stream, boundary] = streamFromMultipart(async (writer) => {
-      await writer.writeFile("test", "test.bin", textBytesReader);
+      await writer.writeFile("test", "test.bin", testBytesReader);
       await writer.writeField("deno", "land");
     });
-
-    const reader = readerFromStreamReader(stream.getReader());
+    const reader = readerFromIterable(stream);
     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);
+    assertEquals(formFile.content, testBytes);
 
     // Ensure the field was serialized correctly.
     assertEquals(form.value("deno"), "land");