feat(std/http): make req.body a Reader (#3575)

std/http/server.ts

@@ -8,12 +8,7 @@ import { BufReader, BufWriter, UnexpectedEOFError } from "../io/bufio.ts";
 import { TextProtoReader } from "../textproto/mod.ts";
 import { STATUS_TEXT } from "./http_status.ts";
 import { assert } from "../testing/asserts.ts";
-import {
+import { deferred, Deferred, MuxAsyncIterator } from "../util/async.ts";
-  collectUint8Arrays,
-  deferred,
-  Deferred,
-  MuxAsyncIterator
-} from "../util/async.ts";
 
 function bufWriter(w: Writer): BufWriter {
   if (w instanceof BufWriter) {
@@ -97,6 +92,17 @@ export async function writeResponse(w: Writer, r: Response): Promise<void> {
   await writer.flush();
 }
 
+export class ServerRequestBody implements Reader {
+  constructor(private it: AsyncIterator<number, undefined, Uint8Array>) {}
+  async read(p: Uint8Array): Promise<number | Deno.EOF> {
+    const res = await this.it.next(p);
+    if (res.done) {
+      return Deno.EOF;
+    }
+    return res.value;
+  }
+}
+
 export class ServerRequest {
   url!: string;
   method!: string;
@@ -109,24 +115,75 @@ export class ServerRequest {
   w!: BufWriter;
   done: Deferred<Error | undefined> = deferred();
 
-  public async *bodyStream(): AsyncIterableIterator<Uint8Array> {
+  private _contentLength: number | undefined | null = undefined;
-    if (this.headers.has("content-length")) {
+  /**
-      const len = +this.headers.get("content-length")!;
+   * Value of Content-Length header.
-      if (Number.isNaN(len)) {
+   * If null, then content length is invalid or not given (e.g. chunked encoding).
-        return new Uint8Array(0);
+   */
+  get contentLength(): number | null {
+    // undefined means not cached.
+    // null means invalid or not provided.
+    if (this._contentLength === undefined) {
+      if (this.headers.has("content-length")) {
+        this._contentLength = +this.headers.get("content-length")!;
+        // Convert NaN to null (as NaN harder to test)
+        if (Number.isNaN(this._contentLength)) {
+          this._contentLength = null;
+        }
+      } else {
+        this._contentLength = null;
+      }
+    }
+    return this._contentLength;
+  }
+
+  private _body: ServerRequestBody | null = null;
+
+  /**
+   * Body of the request.
+   *
+   *     const buf = new Uint8Array(req.contentLength);
+   *     let bufSlice = buf;
+   *     let totRead = 0;
+   *     while (true) {
+   *       const nread = await req.body.read(bufSlice);
+   *       if (nread === Deno.EOF) break;
+   *       totRead += nread;
+   *       if (totRead >= req.contentLength) break;
+   *       bufSlice = bufSlice.subarray(nread);
+   *     }
+   */
+  get body(): ServerRequestBody {
+    if (!this._body) {
+      const stream = this._bodyStream();
+      stream.next(); // drop dummy such that first read is not empty.
+      this._body = new ServerRequestBody(stream);
+    }
+    return this._body;
+  }
+
+  /**
+   * Internal: actually reading body. Each step, buf to use is passed
+   * in through yield result.
+   * Returns on no more data to read or error.
+   */
+  private async *_bodyStream(): AsyncIterator<number, undefined, Uint8Array> {
+    let buf = yield 0; // dummy yield to retrieve user provided buf.
+    if (this.headers.has("content-length")) {
+      const len = this.contentLength;
+      if (len === null) {
+        return;
       }
-      let buf = new Uint8Array(1024);
       let rr = await this.r.read(buf);
       let nread = rr === Deno.EOF ? 0 : rr;
       let nreadTotal = nread;
       while (rr !== Deno.EOF && nreadTotal < len) {
-        yield buf.subarray(0, nread);
+        buf = yield nread;
-        buf = new Uint8Array(1024);
         rr = await this.r.read(buf);
         nread = rr === Deno.EOF ? 0 : rr;
         nreadTotal += nread;
       }
-      yield buf.subarray(0, nread);
+      yield nread;
     } else {
       if (this.headers.has("transfer-encoding")) {
         const transferEncodings = this.headers
@@ -145,11 +202,17 @@ export class ServerRequest {
             throw new Error("Invalid chunk size");
           }
           while (chunkSize > 0) {
-            const data = new Uint8Array(chunkSize);
+            let currChunkOffset = 0;
-            if ((await this.r.readFull(data)) === Deno.EOF) {
+            // Since given readBuffer might be smaller, loop.
-              throw new UnexpectedEOFError();
+            while (currChunkOffset < chunkSize) {
+              // Try to be as large as chunkSize. Might be smaller though.
+              const bufferToFill = buf.subarray(0, chunkSize);
+              if ((await this.r.readFull(bufferToFill)) === Deno.EOF) {
+                throw new UnexpectedEOFError();
+              }
+              currChunkOffset += bufferToFill.length;
+              buf = yield bufferToFill.length;
             }
-            yield data;
             await this.r.readLine(); // Consume \r\n
             line = await tp.readLine();
             if (line === Deno.EOF) throw new UnexpectedEOFError();
@@ -182,16 +245,10 @@ export class ServerRequest {
         }
         // TODO: handle other transfer-encoding types
       }
-      // Otherwise...
+      // Otherwise... Do nothing
-      yield new Uint8Array(0);
     }
   }
 
-  // Read the body of the request into a single Uint8Array
-  public async body(): Promise<Uint8Array> {
-    return collectUint8Arrays(this.bodyStream());
-  }
-
   async respond(r: Response): Promise<void> {
     let err: Error | undefined;
     try {

std/http/server_test.ts

@@ -96,6 +96,40 @@ test(async function responseWrite(): Promise<void> {
   }
 });
 
+test(async function requestContentLength(): Promise<void> {
+  // Has content length
+  {
+    const req = new ServerRequest();
+    req.headers = new Headers();
+    req.headers.set("content-length", "5");
+    const buf = new Buffer(enc.encode("Hello"));
+    req.r = new BufReader(buf);
+    assertEquals(req.contentLength, 5);
+  }
+  // No content length
+  {
+    const shortText = "Hello";
+    const req = new ServerRequest();
+    req.headers = new Headers();
+    req.headers.set("transfer-encoding", "chunked");
+    let chunksData = "";
+    let chunkOffset = 0;
+    const maxChunkSize = 70;
+    while (chunkOffset < shortText.length) {
+      const chunkSize = Math.min(maxChunkSize, shortText.length - chunkOffset);
+      chunksData += `${chunkSize.toString(16)}\r\n${shortText.substr(
+        chunkOffset,
+        chunkSize
+      )}\r\n`;
+      chunkOffset += chunkSize;
+    }
+    chunksData += "0\r\n\r\n";
+    const buf = new Buffer(enc.encode(chunksData));
+    req.r = new BufReader(buf);
+    assertEquals(req.contentLength, null);
+  }
+});
+
 test(async function requestBodyWithContentLength(): Promise<void> {
   {
     const req = new ServerRequest();
@@ -103,7 +137,7 @@ test(async function requestBodyWithContentLength(): Promise<void> {
     req.headers.set("content-length", "5");
     const buf = new Buffer(enc.encode("Hello"));
     req.r = new BufReader(buf);
-    const body = dec.decode(await req.body());
+    const body = dec.decode(await Deno.readAll(req.body));
     assertEquals(body, "Hello");
   }
 
@@ -115,7 +149,7 @@ test(async function requestBodyWithContentLength(): Promise<void> {
     req.headers.set("Content-Length", "5000");
     const buf = new Buffer(enc.encode(longText));
     req.r = new BufReader(buf);
-    const body = dec.decode(await req.body());
+    const body = dec.decode(await Deno.readAll(req.body));
     assertEquals(body, longText);
   }
 });
@@ -140,7 +174,7 @@ test(async function requestBodyWithTransferEncoding(): Promise<void> {
     chunksData += "0\r\n\r\n";
     const buf = new Buffer(enc.encode(chunksData));
     req.r = new BufReader(buf);
-    const body = dec.decode(await req.body());
+    const body = dec.decode(await Deno.readAll(req.body));
     assertEquals(body, shortText);
   }
 
@@ -164,12 +198,12 @@ test(async function requestBodyWithTransferEncoding(): Promise<void> {
     chunksData += "0\r\n\r\n";
     const buf = new Buffer(enc.encode(chunksData));
     req.r = new BufReader(buf);
-    const body = dec.decode(await req.body());
+    const body = dec.decode(await Deno.readAll(req.body));
     assertEquals(body, longText);
   }
 });
 
-test(async function requestBodyStreamWithContentLength(): Promise<void> {
+test(async function requestBodyReaderWithContentLength(): Promise<void> {
   {
     const shortText = "Hello";
     const req = new ServerRequest();
@@ -177,16 +211,20 @@ test(async function requestBodyStreamWithContentLength(): Promise<void> {
     req.headers.set("content-length", "" + shortText.length);
     const buf = new Buffer(enc.encode(shortText));
     req.r = new BufReader(buf);
-    const it = await req.bodyStream();
+    const readBuf = new Uint8Array(6);
     let offset = 0;
-    for await (const chunk of it) {
+    while (offset < shortText.length) {
-      const s = dec.decode(chunk);
+      const nread = await req.body.read(readBuf);
-      assertEquals(shortText.substr(offset, s.length), s);
+      assertNotEOF(nread);
-      offset += s.length;
+      const s = dec.decode(readBuf.subarray(0, nread as number));
+      assertEquals(shortText.substr(offset, nread as number), s);
+      offset += nread as number;
     }
+    const nread = await req.body.read(readBuf);
+    assertEquals(nread, Deno.EOF);
   }
 
-  // Larger than internal buf
+  // Larger than given buf
   {
     const longText = "1234\n".repeat(1000);
     const req = new ServerRequest();
@@ -194,17 +232,21 @@ test(async function requestBodyStreamWithContentLength(): Promise<void> {
     req.headers.set("Content-Length", "5000");
     const buf = new Buffer(enc.encode(longText));
     req.r = new BufReader(buf);
-    const it = await req.bodyStream();
+    const readBuf = new Uint8Array(1000);
     let offset = 0;
-    for await (const chunk of it) {
+    while (offset < longText.length) {
-      const s = dec.decode(chunk);
+      const nread = await req.body.read(readBuf);
-      assertEquals(longText.substr(offset, s.length), s);
+      assertNotEOF(nread);
-      offset += s.length;
+      const s = dec.decode(readBuf.subarray(0, nread as number));
+      assertEquals(longText.substr(offset, nread as number), s);
+      offset += nread as number;
     }
+    const nread = await req.body.read(readBuf);
+    assertEquals(nread, Deno.EOF);
   }
 });
 
-test(async function requestBodyStreamWithTransferEncoding(): Promise<void> {
+test(async function requestBodyReaderWithTransferEncoding(): Promise<void> {
   {
     const shortText = "Hello";
     const req = new ServerRequest();
@@ -224,13 +266,17 @@ test(async function requestBodyStreamWithTransferEncoding(): Promise<void> {
     chunksData += "0\r\n\r\n";
     const buf = new Buffer(enc.encode(chunksData));
     req.r = new BufReader(buf);
-    const it = await req.bodyStream();
+    const readBuf = new Uint8Array(6);
     let offset = 0;
-    for await (const chunk of it) {
+    while (offset < shortText.length) {
-      const s = dec.decode(chunk);
+      const nread = await req.body.read(readBuf);
-      assertEquals(shortText.substr(offset, s.length), s);
+      assertNotEOF(nread);
-      offset += s.length;
+      const s = dec.decode(readBuf.subarray(0, nread as number));
+      assertEquals(shortText.substr(offset, nread as number), s);
+      offset += nread as number;
     }
+    const nread = await req.body.read(readBuf);
+    assertEquals(nread, Deno.EOF);
   }
 
   // Larger than internal buf
@@ -253,13 +299,17 @@ test(async function requestBodyStreamWithTransferEncoding(): Promise<void> {
     chunksData += "0\r\n\r\n";
     const buf = new Buffer(enc.encode(chunksData));
     req.r = new BufReader(buf);
-    const it = await req.bodyStream();
+    const readBuf = new Uint8Array(1000);
     let offset = 0;
-    for await (const chunk of it) {
+    while (offset < longText.length) {
-      const s = dec.decode(chunk);
+      const nread = await req.body.read(readBuf);
-      assertEquals(longText.substr(offset, s.length), s);
+      assertNotEOF(nread);
-      offset += s.length;
+      const s = dec.decode(readBuf.subarray(0, nread as number));
+      assertEquals(longText.substr(offset, nread as number), s);
+      offset += nread as number;
     }
+    const nread = await req.body.read(readBuf);
+    assertEquals(nread, Deno.EOF);
   }
 });
 
@@ -610,7 +660,7 @@ if (Deno.build.os !== "win") {
         for await (const req of server) {
           connRid = req.conn.rid;
           reqCount++;
-          await req.body();
+          await Deno.readAll(req.body);
           await connClosedPromise;
           try {
             await req.respond({