Redesign of http server module (#188)

http/README.md

@@ -5,13 +5,22 @@ A framework for creating HTTP/HTTPS server.
 ## Example
 
 ```typescript
-import { serve } from "https://deno.land/x/http/server.ts";
-const s = serve("0.0.0.0:8000");
+import { createServer } from "https://deno.land/x/http/server.ts";
+import { encode } from "https://deno.land/x/strings/strings.ts";
 
 async function main() {
-  for await (const req of s) {
-    req.respond({ body: new TextEncoder().encode("Hello World\n") });
-  }
+  const server = createServer();
+  server.handle("/", async (req, res) => {
+    await res.respond({
+      status: 200,
+      body: encode("ok")
+    });
+  });
+  server.handle(new RegExp("/foo/(?<id>.+)"), async (req, res) => {
+    const { id } = req.match.groups;
+    await res.respondJson({ id });
+  });
+  server.listen("127.0.0.1:8080");
 }
 
 main();

http/file_server.ts

@@ -10,7 +10,7 @@ import {
   listenAndServe,
   ServerRequest,
   setContentLength,
-  Response
+  ServerResponse
 } from "./server.ts";
 import { cwd, DenoError, ErrorKind, args, stat, readDir, open } from "deno";
 import { extname } from "../fs/path.ts";
@@ -195,14 +195,14 @@ async function serveFallback(req: ServerRequest, e: Error) {
   }
 }
 
-function serverLog(req: ServerRequest, res: Response) {
+function serverLog(req: ServerRequest, res: ServerResponse) {
   const d = new Date().toISOString();
   const dateFmt = `[${d.slice(0, 10)} ${d.slice(11, 19)}]`;
   const s = `${dateFmt} "${req.method} ${req.url} ${req.proto}" ${res.status}`;
   console.log(s);
 }
 
-function setCORS(res: Response) {
+function setCORS(res: ServerResponse) {
   if (!res.headers) {
     res.headers = new Headers();
   }
@@ -213,11 +213,11 @@ function setCORS(res: Response) {
   );
 }
 
-listenAndServe(addr, async req => {
+listenAndServe(addr, async (req, res) => {
   const fileName = req.url.replace(/\/$/, "");
   const filePath = currentDir + fileName;
 
-  let response: Response;
+  let response: ServerResponse;
 
   try {
     const fileInfo = await stat(filePath);
@@ -235,7 +235,7 @@ listenAndServe(addr, async req => {
       setCORS(response);
     }
     serverLog(req, response);
-    req.respond(response);
+    res.respond(response);
   }
 });
 

http/http_bench.ts

@@ -1,6 +1,6 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 import * as deno from "deno";
-import { serve } from "./mod.ts";
+import { serve } from "./server.ts";
 
 const addr = deno.args[1] || "127.0.0.1:4500";
 const server = serve(addr);
@@ -8,8 +8,13 @@ const server = serve(addr);
 const body = new TextEncoder().encode("Hello World");
 
 async function main(): Promise<void> {
+  try {
     for await (const request of server) {
-    await request.respond({ status: 200, body });
+      await request.responder.respond({ status: 200, body });
+    }
+  } catch (e) {
+    console.log(e.stack);
+    console.error(e);
   }
 }
 

http/readers.ts

@@ -0,0 +1,78 @@
+import { Reader, ReadResult } from "deno";
+import { BufReader } from "../io/bufio.ts";
+import { TextProtoReader } from "../textproto/mod.ts";
+import { assert } from "../testing/mod.ts";
+
+export class BodyReader implements Reader {
+  total: number;
+  bufReader: BufReader;
+
+  constructor(reader: Reader, private contentLength: number) {
+    this.total = 0;
+    this.bufReader = new BufReader(reader);
+  }
+
+  async read(p: Uint8Array): Promise<ReadResult> {
+    if (p.length > this.contentLength - this.total) {
+      const buf = new Uint8Array(this.contentLength - this.total);
+      const [nread, err] = await this.bufReader.readFull(buf);
+      if (err && err !== "EOF") {
+        throw err;
+      }
+      p.set(buf);
+      this.total += nread;
+      assert.assert(
+        this.total === this.contentLength,
+        `${this.total}, ${this.contentLength}`
+      );
+      return { nread, eof: true };
+    } else {
+      const { nread } = await this.bufReader.read(p);
+      this.total += nread;
+      return { nread, eof: false };
+    }
+  }
+}
+
+export class ChunkedBodyReader implements Reader {
+  bufReader = new BufReader(this.reader);
+  tpReader = new TextProtoReader(this.bufReader);
+
+  constructor(private reader: Reader) {}
+
+  chunks: Uint8Array[] = [];
+  crlfBuf = new Uint8Array(2);
+  finished: boolean = false;
+
+  async read(p: Uint8Array): Promise<ReadResult> {
+    const [line, sizeErr] = await this.tpReader.readLine();
+    if (sizeErr) {
+      throw sizeErr;
+    }
+    const len = parseInt(line, 16);
+    if (len === 0) {
+      this.finished = true;
+      await this.bufReader.readFull(this.crlfBuf);
+      return { nread: 0, eof: true };
+    } else {
+      const buf = new Uint8Array(len);
+      await this.bufReader.readFull(buf);
+      await this.bufReader.readFull(this.crlfBuf);
+      this.chunks.push(buf);
+    }
+    const buf = this.chunks[0];
+    if (buf) {
+      if (buf.byteLength <= p.byteLength) {
+        p.set(buf);
+        this.chunks.shift();
+        return { nread: buf.byteLength, eof: false };
+      } else {
+        p.set(buf.slice(0, p.byteLength));
+        this.chunks[0] = buf.slice(p.byteLength, buf.byteLength);
+        return { nread: p.byteLength, eof: false };
+      }
+    } else {
+      return { nread: 0, eof: true };
+    }
+  }
+}

http/readers_test.ts

@@ -0,0 +1,12 @@
+import { assert, runTests, test } from "../testing/mod.ts";
+import { ChunkedBodyReader } from "./readers.ts";
+import { StringReader } from "../io/readers.ts";
+import { Buffer, copy } from "deno";
+
+test(async function httpChunkedBodyReader() {
+  const chunked = "3\r\nabc\r\n5\r\ndefgh\r\n0\r\n\r\n";
+  const r = new ChunkedBodyReader(new StringReader(chunked));
+  const w = new Buffer();
+  await copy(w, r);
+  assert.equal(w.toString(), "abcdefgh");
+});

http/server.ts

@@ -1,63 +1,90 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
-import { listen, Conn, toAsyncIterator, Reader, Writer, copy } from "deno";
-import { BufReader, BufState, BufWriter } from "../io/bufio.ts";
+
+import { Conn, copy, listen, Reader, toAsyncIterator, Writer } from "deno";
+import { BufReader, BufWriter } from "../io/bufio.ts";
 import { TextProtoReader } from "../textproto/mod.ts";
 import { STATUS_TEXT } from "./http_status.ts";
 import { assert } from "../testing/mod.ts";
+import { defer, Deferred } from "../util/deferred.ts";
+import { BodyReader, ChunkedBodyReader } from "./readers.ts";
+import { encode } from "../strings/strings.ts";
 
-interface Deferred {
-  promise: Promise<{}>;
-  resolve: () => void;
-  reject: () => void;
+/** basic handler for http request */
+export type HttpHandler = (req: ServerRequest, res: ServerResponder) => unknown;
+
+export type ServerRequest = {
+  /** request path with queries. always begin with / */
+  url: string;
+  /** HTTP method */
+  method: string;
+  /** requested protocol. like HTTP/1.1 */
+  proto: string;
+  /** HTTP Headers */
+  headers: Headers;
+  /** matched result for path pattern  */
+  match: RegExpMatchArray;
+  /** body stream. body with "transfer-encoding: chunked" will automatically be combined into original data */
+  body: Reader;
+};
+
+/** basic responder for http response */
+export interface ServerResponder {
+  respond(response: ServerResponse): Promise<void>;
+
+  respondJson(obj: any, headers?: Headers): Promise<void>;
+
+  respondText(text: string, headers?: Headers): Promise<void>;
+
+  readonly isResponded: boolean;
 }
 
-function deferred(): Deferred {
-  let resolve, reject;
-  const promise = new Promise((res, rej) => {
-    resolve = res;
-    reject = rej;
-  });
-  return {
-    promise,
-    resolve,
-    reject
-  };
+export interface ServerResponse {
+  /**
+   * HTTP status code
+   * @default 200 */
+  status?: number;
+  headers?: Headers;
+  body?: Uint8Array | Reader;
 }
 
 interface ServeEnv {
-  reqQueue: ServerRequest[];
+  reqQueue: { req: ServerRequest; conn: Conn }[];
   serveDeferred: Deferred;
 }
 
 /** Continuously read more requests from conn until EOF
  * Calls maybeHandleReq.
- * bufr is empty on a fresh TCP connection.
- * Would be passed around and reused for later request on same conn
  * TODO: make them async function after this change is done
  * https://github.com/tc39/ecma262/pull/1250
  * See https://v8.dev/blog/fast-async
  */
-function serveConn(env: ServeEnv, conn: Conn, bufr?: BufReader) {
-  readRequest(conn, bufr).then(maybeHandleReq.bind(null, env, conn));
+function serveConn(env: ServeEnv, conn: Conn) {
+  readRequest(conn)
+    .then(maybeHandleReq.bind(null, env, conn))
+    .catch(e => {
+      conn.close();
+    });
 }
 
-function maybeHandleReq(env: ServeEnv, conn: Conn, maybeReq: any) {
-  const [req, _err] = maybeReq;
-  if (_err) {
-    conn.close(); // assume EOF for now...
-    return;
-  }
-  env.reqQueue.push(req); // push req to queue
+function maybeHandleReq(env: ServeEnv, conn: Conn, req: ServerRequest) {
+  env.reqQueue.push({ conn, req }); // push req to queue
   env.serveDeferred.resolve(); // signal while loop to process it
 }
 
-export async function* serve(addr: string) {
+/**
+ * iterate new http request asynchronously
+ * @param addr listening address. like 127.0.0.1:80
+ * @param cancel deferred object for cancellation of serving
+ * */
+export async function* serve(
+  addr: string,
+  cancel: Deferred = defer()
+): AsyncIterableIterator<{ req: ServerRequest; res: ServerResponder }> {
   const listener = listen("tcp", addr);
   const env: ServeEnv = {
     reqQueue: [], // in case multiple promises are ready
-    serveDeferred: deferred()
+    serveDeferred: defer()
   };
-
   // Routine that keeps calling accept
   const acceptRoutine = () => {
     const handleConn = (conn: Conn) => {
@@ -65,47 +92,148 @@ export async function* serve(addr: string) {
       scheduleAccept(); // schedule next accept
     };
     const scheduleAccept = () => {
-      listener.accept().then(handleConn);
+      Promise.race([cancel.promise, listener.accept().then(handleConn)]);
     };
     scheduleAccept();
   };
-
   acceptRoutine();
-
-  // Loop hack to allow yield (yield won't work in callbacks)
   while (true) {
-    await env.serveDeferred.promise;
-    env.serveDeferred = deferred(); // use a new deferred
-    let queueToProcess = env.reqQueue;
+    // do race between accept, serveDeferred and cancel
+    await Promise.race([env.serveDeferred.promise, cancel.promise]);
+    // cancellation deferred resolved
+    if (cancel.handled) {
+      break;
+    }
+    // next serve deferred
+    env.serveDeferred = defer();
+    const queueToProcess = env.reqQueue;
     env.reqQueue = [];
-    for (const result of queueToProcess) {
-      yield result;
-      // Continue read more from conn when user is done with the current req
-      // Moving this here makes it easier to manage
-      serveConn(env, result.conn, result.r);
+    for (const { req, conn } of queueToProcess) {
+      if (req) {
+        const res = createResponder(conn);
+        yield { req, res };
+      }
+      serveConn(env, conn);
     }
   }
   listener.close();
 }
 
-export async function listenAndServe(
-  addr: string,
-  handler: (req: ServerRequest) => void
-) {
+export async function listenAndServe(addr: string, handler: HttpHandler) {
   const server = serve(addr);
 
-  for await (const request of server) {
-    await handler(request);
+  for await (const { req, res } of server) {
+    await handler(req, res);
   }
 }
 
-export interface Response {
-  status?: number;
-  headers?: Headers;
-  body?: Uint8Array | Reader;
+export interface HttpServer {
+  handle(pattern: string | RegExp, handler: HttpHandler);
+
+  listen(addr: string, cancel?: Deferred): Promise<void>;
 }
 
-export function setContentLength(r: Response): void {
+/** create HttpServer object */
+export function createServer(): HttpServer {
+  return new HttpServerImpl();
+}
+
+/** create ServerResponder object */
+export function createResponder(w: Writer): ServerResponder {
+  return new ServerResponderImpl(w);
+}
+
+class HttpServerImpl implements HttpServer {
+  private handlers: { pattern: string | RegExp; handler: HttpHandler }[] = [];
+
+  handle(pattern: string | RegExp, handler: HttpHandler) {
+    this.handlers.push({ pattern, handler });
+  }
+
+  async listen(addr: string, cancel: Deferred = defer()) {
+    for await (const { req, res } of serve(addr, cancel)) {
+      let lastMatch: RegExpMatchArray;
+      let lastHandler: HttpHandler;
+      for (const { pattern, handler } of this.handlers) {
+        const match = req.url.match(pattern);
+        if (!match) {
+          continue;
+        }
+        if (!lastMatch) {
+          lastMatch = match;
+          lastHandler = handler;
+        } else if (match[0].length > lastMatch[0].length) {
+          // use longest match
+          lastMatch = match;
+          lastHandler = handler;
+        }
+      }
+      req.match = lastMatch;
+      if (lastHandler) {
+        await lastHandler(req, res);
+        if (!res.isResponded) {
+          await res.respond({
+            status: 500,
+            body: encode("Not Responded")
+          });
+        }
+      } else {
+        await res.respond({
+          status: 404,
+          body: encode("Not Found")
+        });
+      }
+    }
+  }
+}
+
+class ServerResponderImpl implements ServerResponder {
+  constructor(private w: Writer) {}
+
+  private _responded: boolean = false;
+
+  get isResponded() {
+    return this._responded;
+  }
+
+  private checkIfResponded() {
+    if (this.isResponded) {
+      throw new Error("http: already responded");
+    }
+  }
+
+  respond(response: ServerResponse): Promise<void> {
+    this.checkIfResponded();
+    this._responded = true;
+    return writeResponse(this.w, response);
+  }
+
+  respondJson(obj: any, headers: Headers = new Headers()): Promise<void> {
+    const body = encode(JSON.stringify(obj));
+    if (!headers.has("content-type")) {
+      headers.set("content-type", "application/json");
+    }
+    return this.respond({
+      status: 200,
+      body,
+      headers
+    });
+  }
+
+  respondText(text: string, headers: Headers = new Headers()): Promise<void> {
+    const body = encode(text);
+    if (!headers.has("content-type")) {
+      headers.set("content-type", "text/plain");
+    }
+    return this.respond({
+      status: 200,
+      headers,
+      body
+    });
+  }
+}
+
+export function setContentLength(r: ServerResponse): void {
   if (!r.headers) {
     r.headers = new Headers();
   }
@@ -122,100 +250,6 @@ export function setContentLength(r: Response): void {
   }
 }
 
-export class ServerRequest {
-  url: string;
-  method: string;
-  proto: string;
-  headers: Headers;
-  conn: Conn;
-  r: BufReader;
-  w: BufWriter;
-
-  public async *bodyStream() {
-    if (this.headers.has("content-length")) {
-      const len = +this.headers.get("content-length");
-      if (Number.isNaN(len)) {
-        return new Uint8Array(0);
-      }
-      let buf = new Uint8Array(1024);
-      let rr = await this.r.read(buf);
-      let nread = rr.nread;
-      while (!rr.eof && nread < len) {
-        yield buf.subarray(0, rr.nread);
-        buf = new Uint8Array(1024);
-        rr = await this.r.read(buf);
-        nread += rr.nread;
-      }
-      yield buf.subarray(0, rr.nread);
-    } else {
-      if (this.headers.has("transfer-encoding")) {
-        const transferEncodings = this.headers
-          .get("transfer-encoding")
-          .split(",")
-          .map(e => e.trim().toLowerCase());
-        if (transferEncodings.includes("chunked")) {
-          // Based on https://tools.ietf.org/html/rfc2616#section-19.4.6
-          const tp = new TextProtoReader(this.r);
-          let [line, _] = await tp.readLine();
-          // TODO: handle chunk extension
-          let [chunkSizeString, optExt] = line.split(";");
-          let chunkSize = parseInt(chunkSizeString, 16);
-          if (Number.isNaN(chunkSize) || chunkSize < 0) {
-            throw new Error("Invalid chunk size");
-          }
-          while (chunkSize > 0) {
-            let data = new Uint8Array(chunkSize);
-            let [nread, err] = await this.r.readFull(data);
-            if (nread !== chunkSize) {
-              throw new Error("Chunk data does not match size");
-            }
-            yield data;
-            await this.r.readLine(); // Consume \r\n
-            [line, _] = await tp.readLine();
-            chunkSize = parseInt(line, 16);
-          }
-          const [entityHeaders, err] = await tp.readMIMEHeader();
-          if (!err) {
-            for (let [k, v] of entityHeaders) {
-              this.headers.set(k, v);
-            }
-          }
-          /* Pseudo code from https://tools.ietf.org/html/rfc2616#section-19.4.6
-          length := 0
-          read chunk-size, chunk-extension (if any) and CRLF
-          while (chunk-size > 0) {
-            read chunk-data and CRLF
-            append chunk-data to entity-body
-            length := length + chunk-size
-            read chunk-size and CRLF
-          }
-          read entity-header
-          while (entity-header not empty) {
-            append entity-header to existing header fields
-            read entity-header
-          }
-          Content-Length := length
-          Remove "chunked" from Transfer-Encoding
-          */
-          return; // Must return here to avoid fall through
-        }
-        // TODO: handle other transfer-encoding types
-      }
-      // Otherwise...
-      yield new Uint8Array(0);
-    }
-  }
-
-  // Read the body of the request into a single Uint8Array
-  public async body(): Promise<Uint8Array> {
-    return readAllIterator(this.bodyStream());
-  }
-
-  async respond(r: Response): Promise<void> {
-    return writeResponse(this.w, r);
-  }
-}
-
 function bufWriter(w: Writer): BufWriter {
   if (w instanceof BufWriter) {
     return w;
@@ -224,7 +258,10 @@ function bufWriter(w: Writer): BufWriter {
   }
 }
 
-export async function writeResponse(w: Writer, r: Response): Promise<void> {
+export async function writeResponse(
+  w: Writer,
+  r: ServerResponse
+): Promise<void> {
   const protoMajor = 1;
   const protoMinor = 1;
   const statusCode = r.status || 200;
@@ -282,53 +319,52 @@ async function writeChunkedBody(w: Writer, r: Reader) {
   await writer.write(endChunk);
 }
 
-async function readRequest(
-  c: Conn,
-  bufr?: BufReader
-): Promise<[ServerRequest, BufState]> {
-  if (!bufr) {
-    bufr = new BufReader(c);
-  }
-  const bufw = new BufWriter(c);
-  const req = new ServerRequest();
-  req.conn = c;
-  req.r = bufr!;
-  req.w = bufw;
+export async function readRequest(conn: Reader): Promise<ServerRequest> {
+  const bufr = new BufReader(conn);
   const tp = new TextProtoReader(bufr!);
 
-  let s: string;
-  let err: BufState;
-
   // First line: GET /index.html HTTP/1.0
-  [s, err] = await tp.readLine();
-  if (err) {
-    return [null, err];
+  const [line, lineErr] = await tp.readLine();
+  if (lineErr) {
+    throw lineErr;
   }
-  [req.method, req.url, req.proto] = s.split(" ", 3);
-
-  [req.headers, err] = await tp.readMIMEHeader();
-
-  return [req, err];
+  const [method, url, proto] = line.split(" ", 3);
+  const [headers, headersErr] = await tp.readMIMEHeader();
+  if (headersErr) {
+    throw headersErr;
+  }
+  const contentLength = headers.get("content-length");
+  const body =
+    headers.get("transfer-encoding") === "chunked"
+      ? new ChunkedBodyReader(bufr)
+      : new BodyReader(bufr, parseInt(contentLength));
+  return {
+    method,
+    url,
+    proto,
+    headers,
+    body,
+    match: null
+  };
 }
 
-async function readAllIterator(
-  it: AsyncIterableIterator<Uint8Array>
-): Promise<Uint8Array> {
-  const chunks = [];
-  let len = 0;
-  for await (const chunk of it) {
-    chunks.push(chunk);
-    len += chunk.length;
+export async function readResponse(conn: Reader): Promise<ServerResponse> {
+  const bufr = new BufReader(conn);
+  const tp = new TextProtoReader(bufr!);
+  // First line: HTTP/1,1 200 OK
+  const [line, lineErr] = await tp.readLine();
+  if (lineErr) {
+    throw lineErr;
   }
-  if (chunks.length === 0) {
-    // No need for copy
-    return chunks[0];
+  const [proto, status, statusText] = line.split(" ", 3);
+  const [headers, headersErr] = await tp.readMIMEHeader();
+  if (headersErr) {
+    throw headersErr;
   }
-  const collected = new Uint8Array(len);
-  let offset = 0;
-  for (let chunk of chunks) {
-    collected.set(chunk, offset);
-    offset += chunk.length;
-  }
-  return collected;
+  const contentLength = headers.get("content-length");
+  const body =
+    headers.get("transfer-encoding") === "chunked"
+      ? new ChunkedBodyReader(bufr)
+      : new BodyReader(bufr, parseInt(contentLength));
+  return { status: parseInt(status), headers, body };
 }

http/server_test.ts

@@ -5,19 +5,26 @@
 // Ported from
 // https://github.com/golang/go/blob/master/src/net/http/responsewrite_test.go
 
-import { Buffer } from "deno";
-import { assertEqual, test } from "../testing/mod.ts";
-import { Response, ServerRequest } from "./server.ts";
-import { BufReader, BufWriter } from "../io/bufio.ts";
+import { Buffer, copy, Reader } from "deno";
+import { assert, assertEqual, test } from "../testing/mod.ts";
+import {
+  createResponder,
+  createServer,
+  readRequest,
+  readResponse,
+  ServerResponse,
+  writeResponse
+} from "./server.ts";
+import { encode } from "../strings/strings.ts";
+import { StringReader } from "../io/readers.ts";
+import { StringWriter } from "../io/writers.ts";
+import { defer } from "../util/deferred.ts";
 
 interface ResponseTest {
-  response: Response;
+  response: ServerResponse;
   raw: string;
 }
 
-const enc = new TextEncoder();
-const dec = new TextDecoder();
-
 const responseTests: ResponseTest[] = [
   // Default response
   {
@@ -28,7 +35,7 @@ const responseTests: ResponseTest[] = [
   {
     response: {
       status: 200,
-      body: new Buffer(new TextEncoder().encode("abcdef"))
+      body: new Buffer(encode("abcdef"))
     },
 
     raw:
@@ -38,181 +45,241 @@ const responseTests: ResponseTest[] = [
   }
 ];
 
-test(async function responseWrite() {
-  for (const testCase of responseTests) {
+test(async function httpWriteResponse() {
+  for (const { raw, response } of responseTests) {
     const buf = new Buffer();
-    const bufw = new BufWriter(buf);
-    const request = new ServerRequest();
-    request.w = bufw;
-
-    await request.respond(testCase.response);
-    assertEqual(buf.toString(), testCase.raw);
+    await writeResponse(buf, response);
+    assertEqual(buf.toString(), raw);
   }
 });
 
-test(async function requestBodyWithContentLength() {
-  {
-    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);
-    const body = dec.decode(await req.body());
-    assertEqual(body, "Hello");
-  }
+test(async function httpReadRequest() {
+  const body = "0123456789";
+  const lines = [
+    "GET /index.html?deno=land HTTP/1.1",
+    "Host: deno.land",
+    "Content-Type: text/plain",
+    `Content-Length: ${body.length}`,
+    "\r\n"
+  ];
+  let msg = lines.join("\r\n");
+  msg += body;
+  const req = await readRequest(new StringReader(`${msg}`));
+  assert.equal(req.url, "/index.html?deno=land");
+  assert.equal(req.method, "GET");
+  assert.equal(req.proto, "HTTP/1.1");
+  assert.equal(req.headers.get("host"), "deno.land");
+  assert.equal(req.headers.get("content-type"), "text/plain");
+  assert.equal(req.headers.get("content-length"), `${body.length}`);
+  const w = new StringWriter();
+  await copy(w, req.body);
+  assert.equal(w.toString(), body);
+});
 
-  // Larger than internal buf
+test(async function httpReadRequestChunkedBody() {
+  const lines = [
+    "GET /index.html?deno=land HTTP/1.1",
+    "Host: deno.land",
+    "Content-Type: text/plain",
+    `Transfer-Encoding: chunked`,
+    "\r\n"
+  ];
+  const hd = lines.join("\r\n");
+  const buf = new Buffer();
+  await buf.write(encode(hd));
+  await buf.write(encode("4\r\ndeno\r\n"));
+  await buf.write(encode("5\r\n.land\r\n"));
+  await buf.write(encode("0\r\n\r\n"));
+  const req = await readRequest(buf);
+  assert.equal(req.url, "/index.html?deno=land");
+  assert.equal(req.method, "GET");
+  assert.equal(req.proto, "HTTP/1.1");
+  assert.equal(req.headers.get("host"), "deno.land");
+  assert.equal(req.headers.get("content-type"), "text/plain");
+  assert.equal(req.headers.get("transfer-encoding"), `chunked`);
+  const dest = new Buffer();
+  await copy(dest, req.body);
+  assert.equal(dest.toString(), "deno.land");
+});
+
+test(async function httpServer() {
+  const server = createServer();
+  server.handle("/index", async (req, res) => {
+    await res.respond({
+      status: 200,
+      body: encode("ok")
+    });
+  });
+  server.handle(new RegExp("/foo/(?<id>.+)"), async (req, res) => {
+    const { id } = req.match.groups;
+    await res.respond({
+      status: 200,
+      headers: new Headers({
+        "content-type": "application/json"
+      }),
+      body: encode(JSON.stringify({ id }))
+    });
+  });
+  server.handle("/no-response", async (req, res) => {});
+  const cancel = defer<void>();
+  try {
+    server.listen("127.0.0.1:8080", cancel);
     {
-    const longText = "1234\n".repeat(1000);
-    const req = new ServerRequest();
-    req.headers = new Headers();
-    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());
-    assertEqual(body, longText);
+      const res1 = await fetch("http://127.0.0.1:8080/index");
+      const text = await res1.body.text();
+      assert.equal(res1.status, 200);
+      assert.equal(text, "ok");
+    }
+    {
+      const res2 = await fetch("http://127.0.0.1:8080/foo/123");
+      const json = await res2.body.json();
+      assert.equal(res2.status, 200);
+      assert.equal(res2.headers.get("content-type"), "application/json");
+      assert.equal(json["id"], "123");
+    }
+    {
+      const res = await fetch("http://127.0.0.1:8080/no-response");
+      assert.equal(res.status, 500);
+      const text = await res.body.text();
+      assert.assert(!!text.match("Not Responded"));
+    }
+    {
+      const res = await fetch("http://127.0.0.1:8080/not-found");
+      const text = await res.body.text();
+      assert.equal(res.status, 404);
+      assert.assert(!!text.match("Not Found"));
+    }
+  } finally {
+    cancel.resolve();
   }
 });
 
-test(async function requestBodyWithTransferEncoding() {
-  {
-    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);
-    const body = dec.decode(await req.body());
-    assertEqual(body, shortText);
-  }
-
-  // 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 buf = new Buffer(enc.encode(chunksData));
-    req.r = new BufReader(buf);
-    const body = dec.decode(await req.body());
-    assertEqual(body, longText);
-  }
+test(async function httpServerResponder() {
+  const w = new Buffer();
+  const res = createResponder(w);
+  assert.assert(!res.isResponded);
+  await res.respond({
+    status: 200,
+    headers: new Headers({
+      "content-type": "text/plain"
+    }),
+    body: encode("ok")
+  });
+  assert.assert(res.isResponded);
+  const resp = await readResponse(w);
+  assert.equal(resp.status, 200);
+  assert.equal(resp.headers.get("content-type"), "text/plain");
+  const sw = new StringWriter();
+  await copy(sw, resp.body as Reader);
+  assert.equal(sw.toString(), "ok");
 });
 
-test(async function requestBodyStreamWithContentLength() {
-  {
-    const shortText = "Hello";
-    const req = new ServerRequest();
-    req.headers = new Headers();
-    req.headers.set("content-length", "" + shortText.length);
-    const buf = new Buffer(enc.encode(shortText));
-    req.r = new BufReader(buf);
-    const it = await req.bodyStream();
-    let offset = 0;
-    for await (const chunk of it) {
-      const s = dec.decode(chunk);
-      assertEqual(shortText.substr(offset, s.length), s);
-      offset += s.length;
-    }
-  }
-
-  // Larger than internal buf
-  {
-    const longText = "1234\n".repeat(1000);
-    const req = new ServerRequest();
-    req.headers = new Headers();
-    req.headers.set("Content-Length", "5000");
-    const buf = new Buffer(enc.encode(longText));
-    req.r = new BufReader(buf);
-    const it = await req.bodyStream();
-    let offset = 0;
-    for await (const chunk of it) {
-      const s = dec.decode(chunk);
-      assertEqual(longText.substr(offset, s.length), s);
-      offset += s.length;
-    }
+test(async function httpServerResponderRespondJson() {
+  const w = new Buffer();
+  const res = createResponder(w);
+  const json = {
+    id: 1,
+    deno: {
+      is: "land"
     }
+  };
+  assert.assert(!res.isResponded);
+  await res.respondJson(
+    json,
+    new Headers({
+      deno: "land"
+    })
+  );
+  assert.assert(res.isResponded);
+  const resp = await readResponse(w);
+  assert.equal(resp.status, 200);
+  assert.equal(resp.headers.get("content-type"), "application/json");
+  const sw = new StringWriter();
+  await copy(sw, resp.body as Reader);
+  const resJson = JSON.parse(sw.toString());
+  assert.equal(resJson, json);
+  assert.equal(resp.headers.get("deno"), "land");
 });
 
-test(async function requestBodyStreamWithTransferEncoding() {
-  {
-    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);
-    const it = await req.bodyStream();
-    let offset = 0;
-    for await (const chunk of it) {
-      const s = dec.decode(chunk);
-      assertEqual(shortText.substr(offset, s.length), s);
-      offset += s.length;
-    }
-  }
+test(async function httpServerResponderRespondText() {
+  const w = new Buffer();
+  const res = createResponder(w);
+  assert.assert(!res.isResponded);
+  await res.respondText(
+    "deno.land",
+    new Headers({
+      deno: "land"
+    })
+  );
+  assert.assert(res.isResponded);
+  const resp = await readResponse(w);
+  assert.equal(resp.status, 200);
+  assert.equal(resp.headers.get("content-type"), "text/plain");
+  const sw = new StringWriter();
+  await copy(sw, resp.body as Reader);
+  assert.equal(sw.toString(), "deno.land");
+  assert.equal(resp.headers.get("deno"), "land");
+});
 
-  // Larger than internal buf
+test(async function httpServerResponderShouldThrow() {
+  const w = new Buffer();
   {
-    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;
+    const res = createResponder(w);
+    await res.respond({
+      body: null
+    });
+    await assert.throwsAsync(
+      async () => res.respond({ body: null }),
+      Error,
+      "responded"
+    );
+    await assert.throwsAsync(
+      async () => res.respondJson({}),
+      Error,
+      "responded"
+    );
+    await assert.throwsAsync(
+      async () => res.respondText(""),
+      Error,
+      "responded"
+    );
   }
-    chunksData += "0\r\n\r\n";
-    const buf = new Buffer(enc.encode(chunksData));
-    req.r = new BufReader(buf);
-    const it = await req.bodyStream();
-    let offset = 0;
-    for await (const chunk of it) {
-      const s = dec.decode(chunk);
-      assertEqual(longText.substr(offset, s.length), s);
-      offset += s.length;
+  {
+    const res = createResponder(w);
+    await res.respondText("");
+    await assert.throwsAsync(
+      async () => res.respond({ body: null }),
+      Error,
+      "responded"
+    );
+    await assert.throwsAsync(
+      async () => res.respondJson({}),
+      Error,
+      "responded"
+    );
+    await assert.throwsAsync(
+      async () => res.respondText(""),
+      Error,
+      "responded"
+    );
   }
+  {
+    const res = createResponder(w);
+    await res.respondJson({});
+    await assert.throwsAsync(
+      async () => res.respond({ body: null }),
+      Error,
+      "responded"
+    );
+    await assert.throwsAsync(
+      async () => res.respondJson({}),
+      Error,
+      "responded"
+    );
+    await assert.throwsAsync(
+      async () => res.respondText(""),
+      Error,
+      "responded"
+    );
   }
 });

io/readers_test.ts

@@ -7,9 +7,11 @@ import { decode } from "../strings/strings.ts";
 
 test(async function ioStringReader() {
   const r = new StringReader("abcdef");
-  const { nread, eof } = await r.read(new Uint8Array(6));
+  const buf = new Uint8Array(6);
+  const { nread, eof } = await r.read(buf);
   assert.equal(nread, 6);
   assert.equal(eof, true);
+  assert.equal(decode(buf), "abcdef");
 });
 
 test(async function ioStringReader() {

test.ts

@@ -1,6 +1,8 @@
 #!/usr/bin/env deno -A
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+
 import "benching/test.ts";
+import "util/deferred_test.ts";
 import "colors/test.ts";
 import "datetime/test.ts";
 import "examples/test.ts";
@@ -14,6 +16,7 @@ import "fs/path/test.ts";
 import "io/test.ts";
 import "http/server_test.ts";
 import "http/file_server_test.ts";
+import "http/readers_test.ts";
 import "log/test.ts";
 import "media_types/test.ts";
 import "multipart/formfile_test.ts";

util/deferred.ts

@@ -0,0 +1,42 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+
+export type Deferred<T = any, R = Error> = {
+  promise: Promise<T>;
+  resolve: (t?: T) => void;
+  reject: (r?: R) => void;
+  readonly handled: boolean;
+};
+
+/** Create deferred promise that can be resolved and rejected by outside */
+export function defer<T>(): Deferred<T> {
+  let handled = false;
+  let resolve;
+  let reject;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = r => {
+      handled = true;
+      res(r);
+    };
+    reject = r => {
+      handled = true;
+      rej(r);
+    };
+  });
+  return {
+    promise,
+    resolve,
+    reject,
+    get handled() {
+      return handled;
+    }
+  };
+}
+
+export function isDeferred(x): x is Deferred {
+  return (
+    typeof x === "object" &&
+    x.promise instanceof Promise &&
+    typeof x["resolve"] === "function" &&
+    typeof x["reject"] === "function"
+  );
+}

util/deferred_test.ts

@@ -0,0 +1,16 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+
+import { assert, test } from "../testing/mod.ts";
+import { defer, isDeferred } from "./deferred.ts";
+
+test(async function asyncIsDeferred() {
+  const d = defer();
+  assert.assert(isDeferred(d));
+  assert.assert(
+    isDeferred({
+      promise: null,
+      resolve: () => {},
+      reject: () => {}
+    }) === false
+  );
+});