Consume body on res.respond if user does not call req.body

net/http.ts

@@ -129,8 +129,11 @@ export class ServerRequest {
   r: BufReader;
   w: BufWriter;
 
-  public async *bodyStream() {
-    if (this.headers.has("content-length")) {
+  private _bodyStreamAsyncIter: AsyncIterableIterator<Uint8Array>;
+  private _bodyConsumed: boolean = false;
+
+  private async *_getBodyStreamAsyncIter() {
+    if (this.headers && this.headers.has("content-length")) {
       const len = +this.headers.get("content-length");
       if (Number.isNaN(len)) {
         return new Uint8Array(0);
@@ -145,8 +148,9 @@ export class ServerRequest {
         nread += rr.nread;
       }
       yield buf.subarray(0, rr.nread);
+      this._bodyConsumed = true;
     } else {
-      if (this.headers.has("transfer-encoding")) {
+      if (this.headers && this.headers.has("transfer-encoding")) {
         const transferEncodings = this.headers
           .get("transfer-encoding")
           .split(",")
@@ -195,16 +199,33 @@ export class ServerRequest {
           Content-Length := length
           Remove "chunked" from Transfer-Encoding
           */
+          this._bodyConsumed = true;
           return; // Must return here to avoid fall through
         }
         // TODO: handle other transfer-encoding types
       }
       // Otherwise...
       yield new Uint8Array(0);
+      this._bodyConsumed = true;
     }
   }
 
-  // Read the body of the request into a single Uint8Array
+  /** Get an async iterator on the request body
+   */
+  public bodyStream(): AsyncIterableIterator<Uint8Array> {
+    if (this._bodyConsumed) {
+      throw new Error("Request body has already been consumed");
+    }
+    if (this._bodyStreamAsyncIter) {
+      return this._bodyStreamAsyncIter;
+    }
+    const iter = this._getBodyStreamAsyncIter();
+    this._bodyStreamAsyncIter = iter;
+    return iter;
+  }
+
+  /** Read the body of the request into a single Uint8Array
+   */
   public async body(): Promise<Uint8Array> {
     return readAllIterator(this.bodyStream());
   }
@@ -230,6 +251,11 @@ export class ServerRequest {
   }
 
   async respond(r: Response): Promise<void> {
+    if (!this._bodyConsumed) {
+      await this.body(); // discard body
+      this._bodyConsumed = true;
+    }
+
     const protoMajor = 1;
     const protoMinor = 1;
     const statusCode = r.status || 200;

net/http_test.ts

@@ -225,3 +225,63 @@ test(async function requestBodyStreamWithTransferEncoding() {
     }
   }
 });
+
+test(async function requestRespondWithoutReadingBodyStream() {
+  {
+    const req = new ServerRequest();
+    req.headers = new Headers();
+    req.headers.set("content-length", "5");
+    const rbuf = new Buffer(enc.encode("Hello"));
+    req.r = new BufReader(rbuf);
+    const wbuf = new Buffer(new Uint8Array(1024));
+    req.w = new BufWriter(wbuf);
+    await req.respond({
+      body: enc.encode("World")
+    });
+    // Body should have been secretly consumed by here
+    let err;
+    try {
+      await req.body();
+    } catch (e) {
+      err = e;
+    }
+    assert(!!err);
+    assert(err.message, "Error: Request body has already been consumed");
+  }
+
+  // Larger than internal buf
+  {
+    const longText = "1234\n".repeat(1000);
+    const req = new ServerRequest();
+    req.headers = new Headers();
+    req.headers.set("transfer-encoding", "chunked");
+    let chunksData = "";
+    let chunkOffset = 0;
+    const maxChunkSize = 70;
+    while (chunkOffset < longText.length) {
+      const chunkSize = Math.min(maxChunkSize, longText.length - chunkOffset);
+      chunksData += `${chunkSize.toString(16)}\r\n${longText.substr(
+        chunkOffset,
+        chunkSize
+      )}\r\n`;
+      chunkOffset += chunkSize;
+    }
+    chunksData += "0\r\n\r\n";
+    const rbuf = new Buffer(enc.encode(chunksData));
+    req.r = new BufReader(rbuf);
+    const wbuf = new Buffer(new Uint8Array(1024));
+    req.w = new BufWriter(wbuf);
+    await req.respond({
+      body: enc.encode("World")
+    });
+    // Body should have been secretly consumed by here
+    let err;
+    try {
+      await req.body();
+    } catch (e) {
+      err = e;
+    }
+    assert(!!err);
+    assert(err.message, "Error: Request body has already been consumed");
+  }
+});