Support uploading data from fetch()

Does not yet support streaming, only strings and TypedArrays for now.

js/fetch.ts

@@ -1,5 +1,5 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
-import { assert, createResolvable, notImplemented } from "./util";
+import { assert, createResolvable, notImplemented, isTypedArray } from "./util";
 import * as flatbuffers from "./flatbuffers";
 import { sendAsync } from "./dispatch";
 import * as msg from "gen/msg_generated";
@@ -204,6 +204,7 @@ export async function fetch(
   let url: string;
   let method: string | null = null;
   let headers: domTypes.Headers | null = null;
+  let body: ArrayBufferView | undefined;
 
   if (typeof input === "string") {
     url = input;
@@ -217,6 +218,16 @@ export async function fetch(
       } else {
         headers = null;
       }
+
+      if (init.body) {
+        if (typeof init.body === "string") {
+          body = new TextEncoder().encode(init.body);
+        } else if (isTypedArray(init.body)) {
+          body = init.body;
+        } else {
+          notImplemented();
+        }
+      }
     }
   } else {
     url = input.url;
@@ -232,7 +243,8 @@ export async function fetch(
   const resBase = await sendAsync(
     builder,
     msg.Any.Fetch,
-    msg.Fetch.endFetch(builder)
+    msg.Fetch.endFetch(builder),
+    body
   );
 
   // Decode FetchRes

js/fetch_test.ts

@@ -47,24 +47,36 @@ testPerm({ net: true }, async function responseClone() {
   }
 });
 
-testPerm({ net: true }, async function fetchRequest() {
-  const addr = "127.0.0.1:4501";
+function bufferServer(addr: string): deno.Buffer {
   const listener = deno.listen("tcp", addr);
   const buf = new deno.Buffer();
   listener.accept().then(async conn => {
-    buf.readFrom(conn);
-    await conn.write(
+    const p1 = buf.readFrom(conn);
+    const p2 = conn.write(
       new TextEncoder().encode(
         "HTTP/1.0 404 Not Found\r\nContent-Length: 2\r\n\r\nNF"
       )
     );
+    // Wait for both an EOF on the read side of the socket and for the write to
+    // complete before closing it. Due to keep-alive, the EOF won't be sent
+    // until the Connection close (HTTP/1.0) response, so readFrom() can't
+    // proceed write. Conversely, if readFrom() is async, waiting for the
+    // write() to complete is not a guarantee that we've read the incoming
+    // request.
+    await Promise.all([p1, p2]);
     conn.close();
+    listener.close();
   });
+  return buf;
+}
+
+testPerm({ net: true }, async function fetchRequest() {
+  const addr = "127.0.0.1:4501";
+  const buf = bufferServer(addr);
   const response = await fetch(`http://${addr}/blah`, {
     method: "POST",
     headers: [["Hello", "World"], ["Foo", "Bar"]]
   });
-  listener.close();
   assertEqual(response.status, 404);
   assertEqual(response.headers.get("Content-Length"), "2");
 
@@ -77,3 +89,52 @@ testPerm({ net: true }, async function fetchRequest() {
   ].join("");
   assertEqual(actual, expected);
 });
+
+testPerm({ net: true }, async function fetchPostBodyString() {
+  const addr = "127.0.0.1:4502";
+  const buf = bufferServer(addr);
+  const body = "hello world";
+  const response = await fetch(`http://${addr}/blah`, {
+    method: "POST",
+    headers: [["Hello", "World"], ["Foo", "Bar"]],
+    body
+  });
+  assertEqual(response.status, 404);
+  assertEqual(response.headers.get("Content-Length"), "2");
+
+  const actual = new TextDecoder().decode(buf.bytes());
+  const expected = [
+    "POST /blah HTTP/1.1\r\n",
+    "hello: World\r\n",
+    "foo: Bar\r\n",
+    `host: ${addr}\r\n`,
+    `content-length: ${body.length}\r\n\r\n`,
+    body
+  ].join("");
+  assertEqual(actual, expected);
+});
+
+testPerm({ net: true }, async function fetchPostBodyTypedArray() {
+  const addr = "127.0.0.1:4503";
+  const buf = bufferServer(addr);
+  const bodyStr = "hello world";
+  const body = new TextEncoder().encode(bodyStr);
+  const response = await fetch(`http://${addr}/blah`, {
+    method: "POST",
+    headers: [["Hello", "World"], ["Foo", "Bar"]],
+    body
+  });
+  assertEqual(response.status, 404);
+  assertEqual(response.headers.get("Content-Length"), "2");
+
+  const actual = new TextDecoder().decode(buf.bytes());
+  const expected = [
+    "POST /blah HTTP/1.1\r\n",
+    "hello: World\r\n",
+    "foo: Bar\r\n",
+    `host: ${addr}\r\n`,
+    `content-length: ${body.byteLength}\r\n\r\n`,
+    bodyStr
+  ].join("");
+  assertEqual(actual, expected);
+});

js/util.ts

@@ -125,3 +125,9 @@ export function deferred(): Deferred {
     reject: reject!
   };
 }
+
+// tslint:disable-next-line:variable-name
+const TypedArrayConstructor = Object.getPrototypeOf(Uint8Array);
+export function isTypedArray(x: unknown): x is TypedArray {
+  return x instanceof TypedArrayConstructor;
+}

src/ops.rs

@@ -386,7 +386,6 @@ fn op_fetch(
   base: &msg::Base,
   data: &'static mut [u8],
 ) -> Box<Op> {
-  assert_eq!(data.len(), 0);
   let inner = base.inner_as_fetch().unwrap();
   let cmd_id = base.cmd_id();
 
@@ -394,7 +393,12 @@ fn op_fetch(
   assert!(header.is_request());
   let url = header.url().unwrap();
 
-  let body = hyper::Body::empty();
+  let body = if data.len() == 0 {
+    hyper::Body::empty()
+  } else {
+    hyper::Body::from(Vec::from(data))
+  };
+
   let req = msg_util::deserialize_request(header, body);
 
   if let Err(e) = state.check_net(url) {