io: refactor BufReader/Writer interfaces to be more idiomatic (denoland/deno_std#444)

Thanks Vincent Le Goff (@zekth) for porting over the CSV reader
implementation.

Fixes: denoland/deno_std#436

Original: https://github.com/denoland/deno_std/commit/0ee6334b698072b50c6f5ac8d42d34dc4c94b948

4 files changed, 181 insertions, 134 deletions

diff --git a/http/file_server_test.ts b/http/file_server_test.ts
index 578b0e624..1e2d86c4d 100644
--- a/http/file_server_test.ts
+++ b/http/file_server_test.ts
@@ -3,7 +3,7 @@ const { readFile, run } = Deno;
 
 import { test } from "../testing/mod.ts";
 import { assert, assertEquals } from "../testing/asserts.ts";
-import { BufReader } from "../io/bufio.ts";
+import { BufReader, EOF } from "../io/bufio.ts";
 import { TextProtoReader } from "../textproto/mod.ts";
 
 let fileServer;
@@ -22,10 +22,10 @@ async function startFileServer(): Promise<void> {
   });
   // Once fileServer is ready it will write to its stdout.
   const r = new TextProtoReader(new BufReader(fileServer.stdout));
-  const [s, err] = await r.readLine();
-  assert(err == null);
-  assert(s.includes("server listening"));
+  const s = await r.readLine();
+  assert(s !== EOF && s.includes("server listening"));
 }
+
 function killFileServer(): void {
   fileServer.close();
   fileServer.stdout.close();
diff --git a/http/racing_server_test.ts b/http/racing_server_test.ts
index cdcdca1a7..f98072c16 100644
--- a/http/racing_server_test.ts
+++ b/http/racing_server_test.ts
@@ -1,8 +1,8 @@
 const { dial, run } = Deno;
 
-import { test } from "../testing/mod.ts";
+import { test, runIfMain } from "../testing/mod.ts";
 import { assert, assertEquals } from "../testing/asserts.ts";
-import { BufReader } from "../io/bufio.ts";
+import { BufReader, EOF } from "../io/bufio.ts";
 import { TextProtoReader } from "../textproto/mod.ts";
 
 let server;
@@ -13,9 +13,8 @@ async function startServer(): Promise<void> {
   });
   // Once fileServer is ready it will write to its stdout.
   const r = new TextProtoReader(new BufReader(server.stdout));
-  const [s, err] = await r.readLine();
-  assert(err == null);
-  assert(s.includes("Racing server listening..."));
+  const s = await r.readLine();
+  assert(s !== EOF && s.includes("Racing server listening..."));
 }
 function killServer(): void {
   server.close();
@@ -57,9 +56,10 @@ test(async function serverPipelineRace(): Promise<void> {
   const outLines = output.split("\n");
   // length - 1 to disregard last empty line
   for (let i = 0; i < outLines.length - 1; i++) {
-    const [s, err] = await r.readLine();
-    assert(!err);
+    const s = await r.readLine();
     assertEquals(s, outLines[i]);
   }
   killServer();
 });
+
+runIfMain(import.meta);
diff --git a/http/server.ts b/http/server.ts
index 68a9d8780..bdf48fca3 100644
--- a/http/server.ts
+++ b/http/server.ts
@@ -4,10 +4,10 @@ type Listener = Deno.Listener;
 type Conn = Deno.Conn;
 type Reader = Deno.Reader;
 type Writer = Deno.Writer;
-import { BufReader, BufState, BufWriter } from "../io/bufio.ts";
+import { BufReader, BufWriter, EOF, UnexpectedEOFError } from "../io/bufio.ts";
 import { TextProtoReader } from "../textproto/mod.ts";
 import { STATUS_TEXT } from "./http_status.ts";
-import { assert, fail } from "../testing/asserts.ts";
+import { assert } from "../testing/asserts.ts";
 import {
   collectUint8Arrays,
   deferred,
@@ -134,7 +134,8 @@ export class ServerRequest {
         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();
+          let line = await tp.readLine();
+          if (line === EOF) throw new UnexpectedEOFError();
           // TODO: handle chunk extension
           let [chunkSizeString] = line.split(";");
           let chunkSize = parseInt(chunkSizeString, 16);
@@ -142,18 +143,18 @@ export class ServerRequest {
             throw new Error("Invalid chunk size");
           }
           while (chunkSize > 0) {
-            let data = new Uint8Array(chunkSize);
-            let [nread] = await this.r.readFull(data);
-            if (nread !== chunkSize) {
-              throw new Error("Chunk data does not match size");
+            const data = new Uint8Array(chunkSize);
+            if ((await this.r.readFull(data)) === EOF) {
+              throw new UnexpectedEOFError();
             }
             yield data;
             await this.r.readLine(); // Consume \r\n
-            [line] = await tp.readLine();
+            line = await tp.readLine();
+            if (line === EOF) throw new UnexpectedEOFError();
             chunkSize = parseInt(line, 16);
           }
-          const [entityHeaders, err] = await tp.readMIMEHeader();
-          if (!err) {
+          const entityHeaders = await tp.readMIMEHeader();
+          if (entityHeaders !== EOF) {
             for (let [k, v] of entityHeaders) {
               this.headers.set(k, v);
             }
@@ -220,70 +221,78 @@ function fixLength(req: ServerRequest): void {
 // ParseHTTPVersion parses a HTTP version string.
 // "HTTP/1.0" returns (1, 0, true).
 // Ported from https://github.com/golang/go/blob/f5c43b9/src/net/http/request.go#L766-L792
-export function parseHTTPVersion(vers: string): [number, number, boolean] {
-  const Big = 1000000; // arbitrary upper bound
-  const digitReg = /^\d+$/; // test if string is only digit
-  let major: number;
-  let minor: number;
-
+export function parseHTTPVersion(vers: string): [number, number] {
   switch (vers) {
     case "HTTP/1.1":
-      return [1, 1, true];
+      return [1, 1];
+
     case "HTTP/1.0":
-      return [1, 0, true];
-  }
+      return [1, 0];
 
-  if (!vers.startsWith("HTTP/")) {
-    return [0, 0, false];
-  }
+    default: {
+      const Big = 1000000; // arbitrary upper bound
+      const digitReg = /^\d+$/; // test if string is only digit
+      let major: number;
+      let minor: number;
 
-  const dot = vers.indexOf(".");
-  if (dot < 0) {
-    return [0, 0, false];
-  }
+      if (!vers.startsWith("HTTP/")) {
+        break;
+      }
 
-  let majorStr = vers.substring(vers.indexOf("/") + 1, dot);
-  major = parseInt(majorStr);
-  if (!digitReg.test(majorStr) || isNaN(major) || major < 0 || major > Big) {
-    return [0, 0, false];
-  }
+      const dot = vers.indexOf(".");
+      if (dot < 0) {
+        break;
+      }
+
+      let majorStr = vers.substring(vers.indexOf("/") + 1, dot);
+      major = parseInt(majorStr);
+      if (
+        !digitReg.test(majorStr) ||
+        isNaN(major) ||
+        major < 0 ||
+        major > Big
+      ) {
+        break;
+      }
 
-  let minorStr = vers.substring(dot + 1);
-  minor = parseInt(minorStr);
-  if (!digitReg.test(minorStr) || isNaN(minor) || minor < 0 || minor > Big) {
-    return [0, 0, false];
+      let minorStr = vers.substring(dot + 1);
+      minor = parseInt(minorStr);
+      if (
+        !digitReg.test(minorStr) ||
+        isNaN(minor) ||
+        minor < 0 ||
+        minor > Big
+      ) {
+        break;
+      }
+
+      return [major, minor];
+    }
   }
-  return [major, minor, true];
+
+  throw new Error(`malformed HTTP version ${vers}`);
 }
 
 export async function readRequest(
   bufr: BufReader
-): Promise<[ServerRequest, BufState]> {
+): Promise<ServerRequest | EOF> {
+  const tp = new TextProtoReader(bufr);
+  const firstLine = await tp.readLine(); // e.g. GET /index.html HTTP/1.0
+  if (firstLine === EOF) return EOF;
+  const headers = await tp.readMIMEHeader();
+  if (headers === EOF) throw new UnexpectedEOFError();
+
   const req = new ServerRequest();
   req.r = bufr;
-  const tp = new TextProtoReader(bufr);
-  let err: BufState;
-  // First line: GET /index.html HTTP/1.0
-  let firstLine: string;
-  [firstLine, err] = await tp.readLine();
-  if (err) {
-    return [null, err];
-  }
   [req.method, req.url, req.proto] = firstLine.split(" ", 3);
-
-  let ok: boolean;
-  [req.protoMinor, req.protoMajor, ok] = parseHTTPVersion(req.proto);
-  if (!ok) {
-    throw Error(`malformed HTTP version ${req.proto}`);
-  }
-
-  [req.headers, err] = await tp.readMIMEHeader();
+  [req.protoMinor, req.protoMajor] = parseHTTPVersion(req.proto);
+  req.headers = headers;
   fixLength(req);
   // TODO(zekth) : add parsing of headers eg:
   // rfc: https://tools.ietf.org/html/rfc7230#section-3.3.2
   // A sender MUST NOT send a Content-Length header field in any message
   // that contains a Transfer-Encoding header field.
-  return [req, err];
+  return req;
 }
 
 export class Server implements AsyncIterable<ServerRequest> {
@@ -302,36 +311,39 @@ export class Server implements AsyncIterable<ServerRequest> {
   ): AsyncIterableIterator<ServerRequest> {
     const bufr = new BufReader(conn);
     const w = new BufWriter(conn);
-    let bufStateErr: BufState;
-    let req: ServerRequest;
+    let req: ServerRequest | EOF;
+    let err: Error | undefined;
 
     while (!this.closing) {
       try {
-        [req, bufStateErr] = await readRequest(bufr);
-      } catch (err) {
-        bufStateErr = err;
+        req = await readRequest(bufr);
+      } catch (e) {
+        err = e;
+        break;
+      }
+      if (req === EOF) {
+        break;
       }
-      if (bufStateErr) break;
+
       req.w = w;
       yield req;
+
       // Wait for the request to be processed before we accept a new request on
       // this connection.
       await req.done;
     }
 
-    if (bufStateErr === "EOF") {
+    if (req === EOF) {
       // The connection was gracefully closed.
-    } else if (bufStateErr instanceof Error) {
+    } else if (err) {
       // An error was thrown while parsing request headers.
       await writeResponse(req.w, {
         status: 400,
-        body: new TextEncoder().encode(`${bufStateErr.message}\r\n\r\n`)
+        body: new TextEncoder().encode(`${err.message}\r\n\r\n`)
       });
     } else if (this.closing) {
       // There are more requests incoming but the server is closing.
       // TODO(ry): send a back a HTTP 503 Service Unavailable status.
-    } else {
-      fail(`unexpected BufState: ${bufStateErr}`);
     }
 
     conn.close();
diff --git a/http/server_test.ts b/http/server_test.ts
index fbab0234f..32f12cc40 100644
--- a/http/server_test.ts
+++ b/http/server_test.ts
@@ -7,7 +7,7 @@
 
 const { Buffer } = Deno;
 import { test, runIfMain } from "../testing/mod.ts";
-import { assert, assertEquals } from "../testing/asserts.ts";
+import { assert, assertEquals, assertNotEquals } from "../testing/asserts.ts";
 import {
   Response,
   ServerRequest,
@@ -15,9 +15,20 @@ import {
   readRequest,
   parseHTTPVersion
 } from "./server.ts";
-import { BufReader, BufWriter } from "../io/bufio.ts";
+import {
+  BufReader,
+  BufWriter,
+  EOF,
+  ReadLineResult,
+  UnexpectedEOFError
+} from "../io/bufio.ts";
 import { StringReader } from "../io/readers.ts";
 
+function assertNotEOF<T extends {}>(val: T | EOF): T {
+  assertNotEquals(val, EOF);
+  return val as T;
+}
+
 interface ResponseTest {
   response: Response;
   raw: string;
@@ -247,21 +258,25 @@ test(async function writeUint8ArrayResponse(): Promise<void> {
   const decoder = new TextDecoder("utf-8");
   const reader = new BufReader(buf);
 
-  let line: Uint8Array;
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), "HTTP/1.1 200 OK");
+  let r: ReadLineResult;
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), "HTTP/1.1 200 OK");
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), `content-length: ${shortText.length}`);
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), `content-length: ${shortText.length}`);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(line.byteLength, 0);
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(r.line.byteLength, 0);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), shortText);
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), shortText);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(line.byteLength, 0);
+  const eof = await reader.readLine();
+  assertEquals(eof, EOF);
 });
 
 test(async function writeStringReaderResponse(): Promise<void> {
@@ -276,24 +291,30 @@ test(async function writeStringReaderResponse(): Promise<void> {
   const decoder = new TextDecoder("utf-8");
   const reader = new BufReader(buf);
 
-  let line: Uint8Array;
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), "HTTP/1.1 200 OK");
+  let r: ReadLineResult;
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), "HTTP/1.1 200 OK");
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), "transfer-encoding: chunked");
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), "transfer-encoding: chunked");
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(line.byteLength, 0);
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(r.line.byteLength, 0);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), shortText.length.toString());
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), shortText.length.toString());
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), shortText);
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), shortText);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
-  assertEquals(decoder.decode(line), "0");
+  r = assertNotEOF(await reader.readLine());
+  assertEquals(decoder.decode(r.line), "0");
+  assertEquals(r.more, false);
 });
 
 test(async function readRequestError(): Promise<void> {
@@ -318,19 +339,20 @@ test(async function testReadRequestError(): Promise<void> {
   const testCases = {
     0: {
       in: "GET / HTTP/1.1\r\nheader: foo\r\n\r\n",
-      headers: [{ key: "header", value: "foo" }],
-      err: null
+      headers: [{ key: "header", value: "foo" }]
     },
-    1: { in: "GET / HTTP/1.1\r\nheader:foo\r\n", err: "EOF", headers: [] },
-    2: { in: "", err: "EOF", headers: [] },
+    1: {
+      in: "GET / HTTP/1.1\r\nheader:foo\r\n",
+      err: UnexpectedEOFError
+    },
+    2: { in: "", err: EOF },
     3: {
       in: "HEAD / HTTP/1.1\r\nContent-Length:4\r\n\r\n",
       err: "http: method cannot contain a Content-Length"
     },
     4: {
       in: "HEAD / HTTP/1.1\r\n\r\n",
-      headers: [],
-      err: null
+      headers: []
     },
     // Multiple Content-Length values should either be
     // deduplicated if same or reject otherwise
@@ -348,7 +370,6 @@ test(async function testReadRequestError(): Promise<void> {
     7: {
       in:
         "PUT / HTTP/1.1\r\nContent-Length: 6 \r\nContent-Length: 6\r\nContent-Length:6\r\n\r\nGopher\r\n",
-      err: null,
       headers: [{ key: "Content-Length", value: "6" }]
     },
     8: {
@@ -363,24 +384,28 @@ test(async function testReadRequestError(): Promise<void> {
     // },
     10: {
       in: "HEAD / HTTP/1.1\r\nContent-Length:0\r\nContent-Length: 0\r\n\r\n",
-      headers: [{ key: "Content-Length", value: "0" }],
-      err: null
+      headers: [{ key: "Content-Length", value: "0" }]
     }
   };
   for (const p in testCases) {
     const test = testCases[p];
     const reader = new BufReader(new StringReader(test.in));
-    let _err;
-    if (test.err && test.err != "EOF") {
-      try {
-        await readRequest(reader);
-      } catch (e) {
-        _err = e;
-      }
-      assertEquals(_err.message, test.err);
+    let err;
+    let req;
+    try {
+      req = await readRequest(reader);
+    } catch (e) {
+      err = e;
+    }
+    if (test.err === EOF) {
+      assertEquals(req, EOF);
+    } else if (typeof test.err === "string") {
+      assertEquals(err.message, test.err);
+    } else if (test.err) {
+      assert(err instanceof test.err);
     } else {
-      const [req, err] = await readRequest(reader);
-      assertEquals(test.err, err);
+      assertEquals(err, undefined);
+      assertNotEquals(req, EOF);
       for (const h of test.headers) {
         assertEquals(req.headers.get(h.key), h.value);
       }
@@ -393,21 +418,31 @@ test({
   name: "[http] parseHttpVersion",
   fn(): void {
     const testCases = [
-      { in: "HTTP/0.9", want: [0, 9, true] },
-      { in: "HTTP/1.0", want: [1, 0, true] },
-      { in: "HTTP/1.1", want: [1, 1, true] },
-      { in: "HTTP/3.14", want: [3, 14, true] },
-      { in: "HTTP", want: [0, 0, false] },
-      { in: "HTTP/one.one", want: [0, 0, false] },
-      { in: "HTTP/1.1/", want: [0, 0, false] },
-      { in: "HTTP/-1.0", want: [0, 0, false] },
-      { in: "HTTP/0.-1", want: [0, 0, false] },
-      { in: "HTTP/", want: [0, 0, false] },
-      { in: "HTTP/1,0", want: [0, 0, false] }
+      { in: "HTTP/0.9", want: [0, 9] },
+      { in: "HTTP/1.0", want: [1, 0] },
+      { in: "HTTP/1.1", want: [1, 1] },
+      { in: "HTTP/3.14", want: [3, 14] },
+      { in: "HTTP", err: true },
+      { in: "HTTP/one.one", err: true },
+      { in: "HTTP/1.1/", err: true },
+      { in: "HTTP/-1.0", err: true },
+      { in: "HTTP/0.-1", err: true },
+      { in: "HTTP/", err: true },
+      { in: "HTTP/1,0", err: true }
     ];
     for (const t of testCases) {
-      const r = parseHTTPVersion(t.in);
-      assertEquals(r, t.want, t.in);
+      let r, err;
+      try {
+        r = parseHTTPVersion(t.in);
+      } catch (e) {
+        err = e;
+      }
+      if (t.err) {
+        assert(err instanceof Error, t.in);
+      } else {
+        assertEquals(err, undefined);
+        assertEquals(r, t.want, t.in);
+      }
     }
   }
 });