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: 0ee6334b69

encoding/csv.ts

@@ -2,7 +2,7 @@
 // https://github.com/golang/go/blob/go1.12.5/src/encoding/csv/
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 
-import { BufReader, BufState } from "../io/bufio.ts";
+import { BufReader, EOF } from "../io/bufio.ts";
 import { TextProtoReader } from "../textproto/mod.ts";
 
 const INVALID_RUNE = ["\r", "\n", '"'];
@@ -25,30 +25,29 @@ export interface ParseOptions {
   fieldsPerRecord?: number;
 }
 
-function chkOptions(opt: ParseOptions): Error | null {
+function chkOptions(opt: ParseOptions): void {
   if (
     INVALID_RUNE.includes(opt.comma) ||
     INVALID_RUNE.includes(opt.comment) ||
     opt.comma === opt.comment
   ) {
-    return Error("Invalid Delimiter");
+    throw new Error("Invalid Delimiter");
   }
-  return null;
 }
 
 export async function read(
   Startline: number,
   reader: BufReader,
   opt: ParseOptions = { comma: ",", comment: "#", trimLeadingSpace: false }
-): Promise<[string[], BufState]> {
+): Promise<string[] | EOF> {
   const tp = new TextProtoReader(reader);
-  let err: BufState;
   let line: string;
   let result: string[] = [];
   let lineIndex = Startline;
 
-  [line, err] = await tp.readLine();
+  const r = await tp.readLine();
-
+  if (r === EOF) return EOF;
+  line = r;
   // Normalize \r\n to \n on all input lines.
   if (
     line.length >= 2 &&
@@ -61,12 +60,12 @@ export async function read(
 
   const trimmedLine = line.trimLeft();
   if (trimmedLine.length === 0) {
-    return [[], err];
+    return [];
   }
 
   // line starting with comment character is ignored
   if (opt.comment && trimmedLine[0] === opt.comment) {
-    return [result, err];
+    return [];
   }
 
   result = line.split(opt.comma);
@@ -92,12 +91,9 @@ export async function read(
     }
   );
   if (quoteError) {
-    return [
+    throw new ParseError(Startline, lineIndex, 'bare " in non-quoted-field');
-      [],
-      new ParseError(Startline, lineIndex, 'bare " in non-quoted-field')
-    ];
   }
-  return [result, err];
+  return result;
 }
 
 export async function readAll(
@@ -107,19 +103,18 @@ export async function readAll(
     trimLeadingSpace: false,
     lazyQuotes: false
   }
-): Promise<[string[][], BufState]> {
+): Promise<string[][]> {
   const result: string[][] = [];
   let _nbFields: number;
-  let err: BufState;
   let lineResult: string[];
   let first = true;
   let lineIndex = 0;
-  err = chkOptions(opt);
+  chkOptions(opt);
-  if (err) return [result, err];
 
   for (;;) {
-    [lineResult, err] = await read(lineIndex, reader, opt);
+    const r = await read(lineIndex, reader, opt);
-    if (err) break;
+    if (r === EOF) break;
+    lineResult = r;
     lineIndex++;
     // If fieldsPerRecord is 0, Read sets it to
     // the number of fields in the first record
@@ -136,16 +131,10 @@ export async function readAll(
 
     if (lineResult.length > 0) {
       if (_nbFields && _nbFields !== lineResult.length) {
-        return [
+        throw new ParseError(lineIndex, lineIndex, "wrong number of fields");
-          null,
-          new ParseError(lineIndex, lineIndex, "wrong number of fields")
-        ];
       }
       result.push(lineResult);
     }
   }
-  if (err !== "EOF") {
+  return result;
-    return [result, err];
-  }
-  return [result, null];
 }

encoding/csv_test.ts

@@ -437,20 +437,31 @@ for (const t of testCases) {
       if (t.LazyQuotes) {
         lazyquote = t.LazyQuotes;
       }
-      const actual = await readAll(new BufReader(new StringReader(t.Input)), {
+      let actual;
-        comma: comma,
-        comment: comment,
-        trimLeadingSpace: trim,
-        fieldsPerRecord: fieldsPerRec,
-        lazyQuotes: lazyquote
-      });
       if (t.Error) {
-        assert(!!actual[1]);
+        let err;
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        try {
-        const e: any = actual[1];
+          actual = await readAll(new BufReader(new StringReader(t.Input)), {
-        assertEquals(e.message, t.Error);
+            comma: comma,
+            comment: comment,
+            trimLeadingSpace: trim,
+            fieldsPerRecord: fieldsPerRec,
+            lazyQuotes: lazyquote
+          });
+        } catch (e) {
+          err = e;
+        }
+        assert(err);
+        assertEquals(err.message, t.Error);
       } else {
-        const expected = [t.Output, null];
+        actual = await readAll(new BufReader(new StringReader(t.Input)), {
+          comma: comma,
+          comment: comment,
+          trimLeadingSpace: trim,
+          fieldsPerRecord: fieldsPerRec,
+          lazyQuotes: lazyquote
+        });
+        const expected = t.Output;
         assertEquals(actual, expected);
       }
     }

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();
+  const s = await r.readLine();
-  assert(err == null);
+  assert(s !== EOF && s.includes("server listening"));
-  assert(s.includes("server listening"));
 }
+
 function killFileServer(): void {
   fileServer.close();
   fileServer.stdout.close();

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();
+  const s = await r.readLine();
-  assert(err == null);
+  assert(s !== EOF && s.includes("Racing server listening..."));
-  assert(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();
+    const s = await r.readLine();
-    assert(!err);
     assertEquals(s, outLines[i]);
   }
   killServer();
 });
+
+runIfMain(import.meta);

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);
+            const data = new Uint8Array(chunkSize);
-            let [nread] = await this.r.readFull(data);
+            if ((await this.r.readFull(data)) === EOF) {
-            if (nread !== chunkSize) {
+              throw new UnexpectedEOFError();
-              throw new Error("Chunk data does not match size");
             }
             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();
+          const entityHeaders = await tp.readMIMEHeader();
-          if (!err) {
+          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] {
+export function parseHTTPVersion(vers: string): [number, number] {
-  const Big = 1000000; // arbitrary upper bound
-  const digitReg = /^\d+$/; // test if string is only digit
-  let major: number;
-  let minor: 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];
+
+    default: {
+      const Big = 1000000; // arbitrary upper bound
+      const digitReg = /^\d+$/; // test if string is only digit
+      let major: number;
+      let minor: number;
+
+      if (!vers.startsWith("HTTP/")) {
+        break;
+      }
+
+      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
+      ) {
+        break;
+      }
+
+      return [major, minor];
+    }
   }
 
-  if (!vers.startsWith("HTTP/")) {
+  throw new Error(`malformed HTTP version ${vers}`);
-    return [0, 0, false];
-  }
-
-  const dot = vers.indexOf(".");
-  if (dot < 0) {
-    return [0, 0, false];
-  }
-
-  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];
-  }
-
-  let minorStr = vers.substring(dot + 1);
-  minor = parseInt(minorStr);
-  if (!digitReg.test(minorStr) || isNaN(minor) || minor < 0 || minor > Big) {
-    return [0, 0, false];
-  }
-  return [major, minor, true];
 }
 
 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);
-
+  [req.protoMinor, req.protoMajor] = parseHTTPVersion(req.proto);
-  let ok: boolean;
+  req.headers = headers;
-  [req.protoMinor, req.protoMajor, ok] = parseHTTPVersion(req.proto);
-  if (!ok) {
-    throw Error(`malformed HTTP version ${req.proto}`);
-  }
-
-  [req.headers, err] = await tp.readMIMEHeader();
   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 | EOF;
-    let req: ServerRequest;
+    let err: Error | undefined;
 
     while (!this.closing) {
       try {
-        [req, bufStateErr] = await readRequest(bufr);
+        req = await readRequest(bufr);
-      } catch (err) {
+      } catch (e) {
-        bufStateErr = err;
+        err = e;
+        break;
       }
-      if (bufStateErr) break;
+      if (req === EOF) {
+        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();

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;
+  let r: ReadLineResult;
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), "HTTP/1.1 200 OK");
+  assertEquals(decoder.decode(r.line), "HTTP/1.1 200 OK");
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), `content-length: ${shortText.length}`);
+  assertEquals(decoder.decode(r.line), `content-length: ${shortText.length}`);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(line.byteLength, 0);
+  assertEquals(r.line.byteLength, 0);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), shortText);
+  assertEquals(decoder.decode(r.line), shortText);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  const eof = await reader.readLine();
-  assertEquals(line.byteLength, 0);
+  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;
+  let r: ReadLineResult;
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), "HTTP/1.1 200 OK");
+  assertEquals(decoder.decode(r.line), "HTTP/1.1 200 OK");
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), "transfer-encoding: chunked");
+  assertEquals(decoder.decode(r.line), "transfer-encoding: chunked");
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(line.byteLength, 0);
+  assertEquals(r.line.byteLength, 0);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), shortText.length.toString());
+  assertEquals(decoder.decode(r.line), shortText.length.toString());
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), shortText);
+  assertEquals(decoder.decode(r.line), shortText);
+  assertEquals(r.more, false);
 
-  line = (await reader.readLine())[0];
+  r = assertNotEOF(await reader.readLine());
-  assertEquals(decoder.decode(line), "0");
+  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" }],
+      headers: [{ key: "header", value: "foo" }]
-      err: null
     },
-    1: { in: "GET / HTTP/1.1\r\nheader:foo\r\n", err: "EOF", headers: [] },
+    1: {
-    2: { in: "", err: "EOF", headers: [] },
+      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: [],
+      headers: []
-      err: null
     },
     // 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" }],
+      headers: [{ key: "Content-Length", value: "0" }]
-      err: null
     }
   };
   for (const p in testCases) {
     const test = testCases[p];
     const reader = new BufReader(new StringReader(test.in));
-    let _err;
+    let err;
-    if (test.err && test.err != "EOF") {
+    let req;
-      try {
+    try {
-        await readRequest(reader);
+      req = await readRequest(reader);
-      } catch (e) {
+    } catch (e) {
-        _err = e;
+      err = e;
-      }
+    }
-      assertEquals(_err.message, test.err);
+    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(err, undefined);
-      assertEquals(test.err, err);
+      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/0.9", want: [0, 9] },
-      { in: "HTTP/1.0", want: [1, 0, true] },
+      { in: "HTTP/1.0", want: [1, 0] },
-      { in: "HTTP/1.1", want: [1, 1, true] },
+      { in: "HTTP/1.1", want: [1, 1] },
-      { in: "HTTP/3.14", want: [3, 14, true] },
+      { in: "HTTP/3.14", want: [3, 14] },
-      { in: "HTTP", want: [0, 0, false] },
+      { in: "HTTP", err: true },
-      { in: "HTTP/one.one", want: [0, 0, false] },
+      { in: "HTTP/one.one", err: true },
-      { in: "HTTP/1.1/", want: [0, 0, false] },
+      { in: "HTTP/1.1/", err: true },
-      { in: "HTTP/-1.0", want: [0, 0, false] },
+      { in: "HTTP/-1.0", err: true },
-      { in: "HTTP/0.-1", want: [0, 0, false] },
+      { in: "HTTP/0.-1", err: true },
-      { in: "HTTP/", want: [0, 0, false] },
+      { in: "HTTP/", err: true },
-      { in: "HTTP/1,0", want: [0, 0, false] }
+      { in: "HTTP/1,0", err: true }
     ];
     for (const t of testCases) {
-      const r = parseHTTPVersion(t.in);
+      let r, err;
-      assertEquals(r, t.want, t.in);
+      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);
+      }
     }
   }
 });

io/bufio.ts

@@ -15,13 +15,28 @@ const MAX_CONSECUTIVE_EMPTY_READS = 100;
 const CR = charCode("\r");
 const LF = charCode("\n");
 
-export type BufState =
+export class BufferFullError extends Error {
-  | null
+  name = "BufferFullError";
-  | "EOF"
+  constructor(public partial: Uint8Array) {
-  | "BufferFull"
+    super("Buffer full");
-  | "ShortWrite"
+  }
-  | "NoProgress"
+}
-  | Error;
+
+export class UnexpectedEOFError extends Error {
+  name = "UnexpectedEOFError";
+  constructor() {
+    super("Unexpected EOF");
+  }
+}
+
+export const EOF: unique symbol = Symbol("EOF");
+export type EOF = typeof EOF;
+
+/** Result type returned by of BufReader.readLine(). */
+export interface ReadLineResult {
+  line: Uint8Array;
+  more: boolean;
+}
 
 /** BufReader implements buffering for a Reader object. */
 export class BufReader implements Reader {
@@ -29,9 +44,9 @@ export class BufReader implements Reader {
   private rd: Reader; // Reader provided by caller.
   private r = 0; // buf read position.
   private w = 0; // buf write position.
-  private lastByte: number;
+  private eof = false;
-  private lastCharSize: number;
+  // private lastByte: number;
-  private err: BufState;
+  // private lastCharSize: number;
 
   /** return new BufReader unless r is BufReader */
   static create(r: Reader, size = DEFAULT_BUF_SIZE): BufReader {
@@ -54,12 +69,6 @@ export class BufReader implements Reader {
     return this.w - this.r;
   }
 
-  private _readErr(): BufState {
-    const err = this.err;
-    this.err = null;
-    return err;
-  }
-
   // Reads a new chunk into the buffer.
   private async _fill(): Promise<void> {
     // Slide existing data to beginning.
@@ -75,24 +84,21 @@ export class BufReader implements Reader {
 
     // Read new data: try a limited number of times.
     for (let i = MAX_CONSECUTIVE_EMPTY_READS; i > 0; i--) {
-      let rr: ReadResult;
+      let rr: ReadResult = await this.rd.read(this.buf.subarray(this.w));
-      try {
-        rr = await this.rd.read(this.buf.subarray(this.w));
-      } catch (e) {
-        this.err = e;
-        return;
-      }
       assert(rr.nread >= 0, "negative read");
       this.w += rr.nread;
       if (rr.eof) {
-        this.err = "EOF";
+        this.eof = true;
         return;
       }
       if (rr.nread > 0) {
         return;
       }
     }
-    this.err = "NoProgress";
+
+    throw new Error(
+      `No progress after ${MAX_CONSECUTIVE_EMPTY_READS} read() calls`
+    );
   }
 
   /** Discards any buffered data, resets all state, and switches
@@ -105,108 +111,96 @@ export class BufReader implements Reader {
   private _reset(buf: Uint8Array, rd: Reader): void {
     this.buf = buf;
     this.rd = rd;
-    this.lastByte = -1;
+    this.eof = false;
-    // this.lastRuneSize = -1;
+    // this.lastByte = -1;
+    // this.lastCharSize = -1;
   }
 
   /** reads data into p.
    * It returns the number of bytes read into p.
    * The bytes are taken from at most one Read on the underlying Reader,
    * hence n may be less than len(p).
-   * At EOF, the count will be zero and err will be io.EOF.
    * To read exactly len(p) bytes, use io.ReadFull(b, p).
    */
   async read(p: Uint8Array): Promise<ReadResult> {
     let rr: ReadResult = { nread: p.byteLength, eof: false };
-    if (rr.nread === 0) {
+    if (p.byteLength === 0) return rr;
-      if (this.err) {
-        throw this._readErr();
-      }
-      return rr;
-    }
 
     if (this.r === this.w) {
-      if (this.err) {
-        throw this._readErr();
-      }
       if (p.byteLength >= this.buf.byteLength) {
         // Large read, empty buffer.
         // Read directly into p to avoid copy.
-        rr = await this.rd.read(p);
+        const rr = await this.rd.read(p);
         assert(rr.nread >= 0, "negative read");
-        if (rr.nread > 0) {
+        // if (rr.nread > 0) {
-          this.lastByte = p[rr.nread - 1];
+        //   this.lastByte = p[rr.nread - 1];
-          // this.lastRuneSize = -1;
+        //   this.lastCharSize = -1;
-        }
+        // }
-        if (this.err) {
-          throw this._readErr();
-        }
         return rr;
       }
+
       // One read.
       // Do not use this.fill, which will loop.
       this.r = 0;
       this.w = 0;
-      try {
+      rr = await this.rd.read(this.buf);
-        rr = await this.rd.read(this.buf);
-      } catch (e) {
-        this.err = e;
-      }
       assert(rr.nread >= 0, "negative read");
-      if (rr.nread === 0) {
+      if (rr.nread === 0) return rr;
-        if (this.err) {
-          throw this._readErr();
-        }
-        return rr;
-      }
       this.w += rr.nread;
     }
 
     // copy as much as we can
-    rr.nread = copyBytes(p as Uint8Array, this.buf.subarray(this.r, this.w), 0);
+    rr.nread = copyBytes(p, this.buf.subarray(this.r, this.w), 0);
     this.r += rr.nread;
-    this.lastByte = this.buf[this.r - 1];
+    // this.lastByte = this.buf[this.r - 1];
-    // this.lastRuneSize = -1;
+    // this.lastCharSize = -1;
     return rr;
   }
 
-  /** reads exactly len(p) bytes into p.
+  /** reads exactly `p.length` bytes into `p`.
+   *
+   * If successful, `p` is returned.
+   *
+   * If the end of the underlying stream has been reached, and there are no more
+   * bytes available in the buffer, `readFull()` returns `EOF` instead.
+   *
+   * An error is thrown if some bytes could be read, but not enough to fill `p`
+   * entirely before the underlying stream reported an error or EOF. Any error
+   * thrown will have a `partial` property that indicates the slice of the
+   * buffer that has been successfully filled with data.
+   *
    * Ported from https://golang.org/pkg/io/#ReadFull
-   * It returns the number of bytes copied and an error if fewer bytes were read.
-   * The error is EOF only if no bytes were read.
-   * If an EOF happens after reading some but not all the bytes,
-   * readFull returns ErrUnexpectedEOF. ("EOF" for current impl)
-   * On return, n == len(p) if and only if err == nil.
-   * If r returns an error having read at least len(buf) bytes,
-   * the error is dropped.
    */
-  async readFull(p: Uint8Array): Promise<[number, BufState]> {
+  async readFull(p: Uint8Array): Promise<Uint8Array | EOF> {
-    let rr = await this.read(p);
+    let bytesRead = 0;
-    let nread = rr.nread;
+    while (bytesRead < p.length) {
-    if (rr.eof) {
+      try {
-      return [nread, nread < p.length ? "EOF" : null];
+        const rr = await this.read(p.subarray(bytesRead));
+        bytesRead += rr.nread;
+        if (rr.eof) {
+          if (bytesRead === 0) {
+            return EOF;
+          } else {
+            throw new UnexpectedEOFError();
+          }
+        }
+      } catch (err) {
+        err.partial = p.subarray(0, bytesRead);
+        throw err;
+      }
     }
-    while (!rr.eof && nread < p.length) {
+    return p;
-      rr = await this.read(p.subarray(nread));
-      nread += rr.nread;
-    }
-    return [nread, nread < p.length ? "EOF" : null];
   }
 
   /** Returns the next byte [0, 255] or -1 if EOF. */
   async readByte(): Promise<number> {
     while (this.r === this.w) {
+      if (this.eof) return -1;
       await this._fill(); // buffer is empty.
-      if (this.err == "EOF") {
-        return -1;
-      }
-      if (this.err != null) {
-        throw this._readErr();
-      }
     }
     const c = this.buf[this.r];
     this.r++;
-    this.lastByte = c;
+    // this.lastByte = c;
     return c;
   }
 
@@ -218,46 +212,73 @@ export class BufReader implements Reader {
    * delim.
    * For simple uses, a Scanner may be more convenient.
    */
-  async readString(_delim: string): Promise<string> {
+  async readString(_delim: string): Promise<string | EOF> {
     throw new Error("Not implemented");
   }
 
-  /** readLine() is a low-level line-reading primitive. Most callers should use
+  /** `readLine()` is a low-level line-reading primitive. Most callers should
-   * readBytes('\n') or readString('\n') instead or use a Scanner.
+   * use `readString('\n')` instead or use a Scanner.
    *
-   * readLine tries to return a single line, not including the end-of-line bytes.
+   * `readLine()` tries to return a single line, not including the end-of-line
-   * If the line was too long for the buffer then isPrefix is set and the
+   * bytes. If the line was too long for the buffer then `more` is set and the
    * beginning of the line is returned. The rest of the line will be returned
-   * from future calls. isPrefix will be false when returning the last fragment
+   * from future calls. `more` will be false when returning the last fragment
    * of the line. The returned buffer is only valid until the next call to
-   * ReadLine. ReadLine either returns a non-nil line or it returns an error,
+   * `readLine()`.
-   * never both.
    *
-   * The text returned from ReadLine does not include the line end ("\r\n" or "\n").
+   * The text returned from ReadLine does not include the line end ("\r\n" or
-   * No indication or error is given if the input ends without a final line end.
+   * "\n").
-   * Calling UnreadByte after ReadLine will always unread the last byte read
+   *
-   * (possibly a character belonging to the line end) even if that byte is not
+   * When the end of the underlying stream is reached, the final bytes in the
-   * part of the line returned by ReadLine.
+   * stream are returned. No indication or error is given if the input ends
+   * without a final line end. When there are no more trailing bytes to read,
+   * `readLine()` returns the `EOF` symbol.
+   *
+   * Calling `unreadByte()` after `readLine()` will always unread the last byte
+   * read (possibly a character belonging to the line end) even if that byte is
+   * not part of the line returned by `readLine()`.
    */
-  async readLine(): Promise<[Uint8Array, boolean, BufState]> {
+  async readLine(): Promise<ReadLineResult | EOF> {
-    let [line, err] = await this.readSlice(LF);
+    let line: Uint8Array | EOF;
+
+    try {
+      line = await this.readSlice(LF);
+    } catch (err) {
+      let { partial } = err;
+      assert(
+        partial instanceof Uint8Array,
+        "bufio: caught error from `readSlice()` without `partial` property"
+      );
+
+      // Don't throw if `readSlice()` failed with `BufferFullError`, instead we
+      // just return whatever is available and set the `more` flag.
+      if (!(err instanceof BufferFullError)) {
+        throw err;
+      }
 
-    if (err === "BufferFull") {
       // Handle the case where "\r\n" straddles the buffer.
-      if (line.byteLength > 0 && line[line.byteLength - 1] === CR) {
+      if (
+        !this.eof &&
+        partial.byteLength > 0 &&
+        partial[partial.byteLength - 1] === CR
+      ) {
         // Put the '\r' back on buf and drop it from line.
         // Let the next call to ReadLine check for "\r\n".
         assert(this.r > 0, "bufio: tried to rewind past start of buffer");
         this.r--;
-        line = line.subarray(0, line.byteLength - 1);
+        partial = partial.subarray(0, partial.byteLength - 1);
       }
-      return [line, true, null];
+
+      return { line: partial, more: !this.eof };
+    }
+
+    if (line === EOF) {
+      return EOF;
     }
 
     if (line.byteLength === 0) {
-      return [line, false, err];
+      return { line, more: false };
     }
-    err = null;
 
     if (line[line.byteLength - 1] == LF) {
       let drop = 1;
@@ -266,98 +287,112 @@ export class BufReader implements Reader {
       }
       line = line.subarray(0, line.byteLength - drop);
     }
-    return [line, false, err];
+    return { line, more: false };
   }
 
-  /** readSlice() reads until the first occurrence of delim in the input,
+  /** `readSlice()` reads until the first occurrence of `delim` in the input,
    * returning a slice pointing at the bytes in the buffer. The bytes stop
-   * being valid at the next read. If readSlice() encounters an error before
+   * being valid at the next read.
-   * finding a delimiter, it returns all the data in the buffer and the error
+   *
-   * itself (often io.EOF).  readSlice() fails with error ErrBufferFull if the
+   * If `readSlice()` encounters an error before finding a delimiter, or the
-   * buffer fills without a delim. Because the data returned from readSlice()
+   * buffer fills without finding a delimiter, it throws an error with a
-   * will be overwritten by the next I/O operation, most clients should use
+   * `partial` property that contains the entire buffer.
-   * readBytes() or readString() instead. readSlice() returns err != nil if and
+   *
-   * only if line does not end in delim.
+   * If `readSlice()` encounters the end of the underlying stream and there are
+   * any bytes left in the buffer, the rest of the buffer is returned. In other
+   * words, EOF is always treated as a delimiter. Once the buffer is empty,
+   * it returns `EOF`.
+   *
+   * Because the data returned from `readSlice()` will be overwritten by the
+   * next I/O operation, most clients should use `readString()` instead.
    */
-  async readSlice(delim: number): Promise<[Uint8Array, BufState]> {
+  async readSlice(delim: number): Promise<Uint8Array | EOF> {
     let s = 0; // search start index
-    let line: Uint8Array;
+    let slice: Uint8Array;
-    let err: BufState;
+
     while (true) {
       // Search buffer.
       let i = this.buf.subarray(this.r + s, this.w).indexOf(delim);
       if (i >= 0) {
         i += s;
-        line = this.buf.subarray(this.r, this.r + i + 1);
+        slice = this.buf.subarray(this.r, this.r + i + 1);
         this.r += i + 1;
         break;
       }
 
-      // Pending error?
+      // EOF?
-      if (this.err) {
+      if (this.eof) {
-        line = this.buf.subarray(this.r, this.w);
+        if (this.r === this.w) {
+          return EOF;
+        }
+        slice = this.buf.subarray(this.r, this.w);
         this.r = this.w;
-        err = this._readErr();
         break;
       }
 
       // Buffer full?
       if (this.buffered() >= this.buf.byteLength) {
         this.r = this.w;
-        line = this.buf;
+        throw new BufferFullError(this.buf);
-        err = "BufferFull";
-        break;
       }
 
       s = this.w - this.r; // do not rescan area we scanned before
 
-      await this._fill(); // buffer is not full
+      // Buffer is not full.
+      try {
+        await this._fill();
+      } catch (err) {
+        err.partial = slice;
+        throw err;
+      }
     }
 
     // Handle last byte, if any.
-    let i = line.byteLength - 1;
+    // const i = slice.byteLength - 1;
-    if (i >= 0) {
+    // if (i >= 0) {
-      this.lastByte = line[i];
+    //   this.lastByte = slice[i];
-      // this.lastRuneSize = -1
+    //   this.lastCharSize = -1
-    }
+    // }
 
-    return [line, err];
+    return slice;
   }
 
-  /** Peek returns the next n bytes without advancing the reader. The bytes stop
+  /** `peek()` returns the next `n` bytes without advancing the reader. The
-   * being valid at the next read call. If Peek returns fewer than n bytes, it
+   * bytes stop being valid at the next read call.
-   * also returns an error explaining why the read is short. The error is
+   *
-   * ErrBufferFull if n is larger than b's buffer size.
+   * When the end of the underlying stream is reached, but there are unread
+   * bytes left in the buffer, those bytes are returned. If there are no bytes
+   * left in the buffer, it returns `EOF`.
+   *
+   * If an error is encountered before `n` bytes are available, `peek()` throws
+   * an error with the `partial` property set to a slice of the buffer that
+   * contains the bytes that were available before the error occurred.
    */
-  async peek(n: number): Promise<[Uint8Array, BufState]> {
+  async peek(n: number): Promise<Uint8Array | EOF> {
     if (n < 0) {
       throw Error("negative count");
     }
 
-    while (
-      this.w - this.r < n &&
-      this.w - this.r < this.buf.byteLength &&
-      this.err == null
-    ) {
-      await this._fill(); // this.w - this.r < len(this.buf) => buffer is not full
-    }
-
-    if (n > this.buf.byteLength) {
-      return [this.buf.subarray(this.r, this.w), "BufferFull"];
-    }
-
-    // 0 <= n <= len(this.buf)
-    let err: BufState;
     let avail = this.w - this.r;
-    if (avail < n) {
+    while (avail < n && avail < this.buf.byteLength && !this.eof) {
-      // not enough data in buffer
+      try {
-      n = avail;
+        await this._fill();
-      err = this._readErr();
+      } catch (err) {
-      if (!err) {
+        err.partial = this.buf.subarray(this.r, this.w);
-        err = "BufferFull";
+        throw err;
       }
+      avail = this.w - this.r;
     }
-    return [this.buf.subarray(this.r, this.r + n), err];
+
+    if (avail === 0 && this.eof) {
+      return EOF;
+    } else if (avail < n && this.eof) {
+      return this.buf.subarray(this.r, this.r + avail);
+    } else if (avail < n) {
+      throw new BufferFullError(this.buf.subarray(this.r, this.w));
+    }
+
+    return this.buf.subarray(this.r, this.r + n);
   }
 }
 
@@ -371,7 +406,7 @@ export class BufReader implements Reader {
 export class BufWriter implements Writer {
   buf: Uint8Array;
   n: number = 0;
-  err: null | BufState = null;
+  err: Error | null = null;
 
   /** return new BufWriter unless w is BufWriter */
   static create(w: Writer, size = DEFAULT_BUF_SIZE): BufWriter {
@@ -400,34 +435,27 @@ export class BufWriter implements Writer {
   }
 
   /** Flush writes any buffered data to the underlying io.Writer. */
-  async flush(): Promise<BufState> {
+  async flush(): Promise<void> {
-    if (this.err != null) {
+    if (this.err !== null) throw this.err;
-      return this.err;
+    if (this.n === 0) return;
-    }
-    if (this.n == 0) {
-      return null;
-    }
 
     let n: number;
-    let err: BufState = null;
     try {
       n = await this.wr.write(this.buf.subarray(0, this.n));
     } catch (e) {
-      err = e;
+      this.err = e;
+      throw e;
     }
 
-    if (n < this.n && err == null) {
+    if (n < this.n) {
-      err = "ShortWrite";
+      if (n > 0) {
-    }
-
-    if (err != null) {
-      if (n > 0 && n < this.n) {
         this.buf.copyWithin(0, n, this.n);
+        this.n -= n;
       }
-      this.n -= n;
+      this.err = new Error("Short write");
-      this.err = err;
+      throw this.err;
-      return err;
     }
+
     this.n = 0;
   }
 
@@ -447,16 +475,20 @@ export class BufWriter implements Writer {
    * Returns the number of bytes written.
    */
   async write(p: Uint8Array): Promise<number> {
+    if (this.err !== null) throw this.err;
+    if (p.length === 0) return 0;
+
     let nn = 0;
     let n: number;
-    while (p.byteLength > this.available() && !this.err) {
+    while (p.byteLength > this.available()) {
-      if (this.buffered() == 0) {
+      if (this.buffered() === 0) {
         // Large write, empty buffer.
         // Write directly from p to avoid copy.
         try {
           n = await this.wr.write(p);
         } catch (e) {
           this.err = e;
+          throw e;
         }
       } else {
         n = copyBytes(this.buf, p, this.n);
@@ -466,9 +498,7 @@ export class BufWriter implements Writer {
       nn += n;
       p = p.subarray(n);
     }
-    if (this.err) {
+
-      throw this.err;
-    }
     n = copyBytes(this.buf, p, this.n);
     this.n += n;
     nn += n;

io/bufio_test.ts

@@ -6,14 +6,30 @@
 const { Buffer } = Deno;
 type Reader = Deno.Reader;
 type ReadResult = Deno.ReadResult;
-import { test } from "../testing/mod.ts";
+import { test, runIfMain } from "../testing/mod.ts";
-import { assert, assertEquals } from "../testing/asserts.ts";
+import {
-import { BufReader, BufWriter } from "./bufio.ts";
+  assert,
+  assertEquals,
+  assertNotEquals,
+  fail
+} from "../testing/asserts.ts";
+import {
+  BufReader,
+  BufWriter,
+  EOF,
+  BufferFullError,
+  UnexpectedEOFError
+} from "./bufio.ts";
 import * as iotest from "./iotest.ts";
 import { charCode, copyBytes, stringsReader } from "./util.ts";
 
 const encoder = new TextEncoder();
 
+function assertNotEOF<T extends {}>(val: T | EOF): T {
+  assertNotEquals(val, EOF);
+  return val as T;
+}
+
 async function readBytes(buf: BufReader): Promise<string> {
   const b = new Uint8Array(1000);
   let nb = 0;
@@ -129,17 +145,20 @@ test(async function bufioBufferFull(): Promise<void> {
   const longString =
     "And now, hello, world! It is the time for all good men to come to the aid of their party";
   const buf = new BufReader(stringsReader(longString), MIN_READ_BUFFER_SIZE);
-  let [line, err] = await buf.readSlice(charCode("!"));
-
   const decoder = new TextDecoder();
-  let actual = decoder.decode(line);
-  assertEquals(err, "BufferFull");
-  assertEquals(actual, "And now, hello, ");
 
-  [line, err] = await buf.readSlice(charCode("!"));
+  try {
-  actual = decoder.decode(line);
+    await buf.readSlice(charCode("!"));
+    fail("readSlice should throw");
+  } catch (err) {
+    assert(err instanceof BufferFullError);
+    assert(err.partial instanceof Uint8Array);
+    assertEquals(decoder.decode(err.partial), "And now, hello, ");
+  }
+
+  const line = assertNotEOF(await buf.readSlice(charCode("!")));
+  const actual = decoder.decode(line);
   assertEquals(actual, "world!");
-  assert(err == null);
 });
 
 const testInput = encoder.encode(
@@ -178,14 +197,12 @@ async function testReadLine(input: Uint8Array): Promise<void> {
     let reader = new TestReader(input, stride);
     let l = new BufReader(reader, input.byteLength + 1);
     while (true) {
-      let [line, isPrefix, err] = await l.readLine();
+      const r = await l.readLine();
-      if (line.byteLength > 0 && err != null) {
+      if (r === EOF) {
-        throw Error("readLine returned both data and error");
-      }
-      assertEquals(isPrefix, false);
-      if (err == "EOF") {
         break;
       }
+      const { line, more } = r;
+      assertEquals(more, false);
       // eslint-disable-next-line @typescript-eslint/restrict-plus-operands
       let want = testOutput.subarray(done, done + line.byteLength);
       assertEquals(
@@ -218,56 +235,51 @@ test(async function bufioPeek(): Promise<void> {
     MIN_READ_BUFFER_SIZE
   );
 
-  let [actual, err] = await buf.peek(1);
+  let actual = assertNotEOF(await buf.peek(1));
   assertEquals(decoder.decode(actual), "a");
-  assert(err == null);
 
-  [actual, err] = await buf.peek(4);
+  actual = assertNotEOF(await buf.peek(4));
   assertEquals(decoder.decode(actual), "abcd");
-  assert(err == null);
 
-  [actual, err] = await buf.peek(32);
+  try {
-  assertEquals(decoder.decode(actual), "abcdefghijklmnop");
+    await buf.peek(32);
-  assertEquals(err, "BufferFull");
+    fail("peek() should throw");
+  } catch (err) {
+    assert(err instanceof BufferFullError);
+    assert(err.partial instanceof Uint8Array);
+    assertEquals(decoder.decode(err.partial), "abcdefghijklmnop");
+  }
 
   await buf.read(p.subarray(0, 3));
   assertEquals(decoder.decode(p.subarray(0, 3)), "abc");
 
-  [actual, err] = await buf.peek(1);
+  actual = assertNotEOF(await buf.peek(1));
   assertEquals(decoder.decode(actual), "d");
-  assert(err == null);
 
-  [actual, err] = await buf.peek(1);
+  actual = assertNotEOF(await buf.peek(1));
   assertEquals(decoder.decode(actual), "d");
-  assert(err == null);
 
-  [actual, err] = await buf.peek(1);
+  actual = assertNotEOF(await buf.peek(1));
   assertEquals(decoder.decode(actual), "d");
-  assert(err == null);
 
-  [actual, err] = await buf.peek(2);
+  actual = assertNotEOF(await buf.peek(2));
   assertEquals(decoder.decode(actual), "de");
-  assert(err == null);
 
   let { eof } = await buf.read(p.subarray(0, 3));
   assertEquals(decoder.decode(p.subarray(0, 3)), "def");
   assert(!eof);
-  assert(err == null);
 
-  [actual, err] = await buf.peek(4);
+  actual = assertNotEOF(await buf.peek(4));
   assertEquals(decoder.decode(actual), "ghij");
-  assert(err == null);
 
   await buf.read(p);
   assertEquals(decoder.decode(p), "ghijklmnop");
 
-  [actual, err] = await buf.peek(0);
+  actual = assertNotEOF(await buf.peek(0));
   assertEquals(decoder.decode(actual), "");
-  assert(err == null);
 
-  [actual, err] = await buf.peek(1);
+  const r = await buf.peek(1);
-  assertEquals(decoder.decode(actual), "");
+  assert(r === EOF);
-  assert(err == "EOF");
   /* TODO
 	// Test for issue 3022, not exposing a reader's error on a successful Peek.
 	buf = NewReaderSize(dataAndEOFReader("abcd"), 32)
@@ -328,16 +340,22 @@ test(async function bufReaderReadFull(): Promise<void> {
   const bufr = new BufReader(data, 3);
   {
     const buf = new Uint8Array(6);
-    const [nread, err] = await bufr.readFull(buf);
+    const r = assertNotEOF(await bufr.readFull(buf));
-    assertEquals(nread, 6);
+    assertEquals(r, buf);
-    assert(!err);
     assertEquals(dec.decode(buf), "Hello ");
   }
   {
     const buf = new Uint8Array(6);
-    const [nread, err] = await bufr.readFull(buf);
+    try {
-    assertEquals(nread, 5);
+      await bufr.readFull(buf);
-    assertEquals(err, "EOF");
+      fail("readFull() should throw");
-    assertEquals(dec.decode(buf.subarray(0, 5)), "World");
+    } catch (err) {
+      assert(err instanceof UnexpectedEOFError);
+      assert(err.partial instanceof Uint8Array);
+      assertEquals(err.partial.length, 5);
+      assertEquals(dec.decode(buf.subarray(0, 5)), "World");
+    }
   }
 });
+
+runIfMain(import.meta);

mime/multipart.ts

@@ -1,19 +1,21 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 
 const { Buffer, copy, remove } = Deno;
+const { min, max } = Math;
 type Closer = Deno.Closer;
 type Reader = Deno.Reader;
 type ReadResult = Deno.ReadResult;
 type Writer = Deno.Writer;
 import { FormFile } from "../multipart/formfile.ts";
-import * as bytes from "../bytes/mod.ts";
+import { equal, findIndex, findLastIndex, hasPrefix } from "../bytes/mod.ts";
+import { extname } from "../fs/path.ts";
 import { copyN } from "../io/ioutil.ts";
 import { MultiReader } from "../io/readers.ts";
 import { tempFile } from "../io/util.ts";
-import { BufReader, BufState, BufWriter } from "../io/bufio.ts";
+import { BufReader, BufWriter, EOF, UnexpectedEOFError } from "../io/bufio.ts";
-import { TextProtoReader } from "../textproto/mod.ts";
 import { encoder } from "../strings/mod.ts";
-import * as path from "../fs/path.ts";
+import { assertStrictEq } from "../testing/asserts.ts";
+import { TextProtoReader } from "../textproto/mod.ts";
 
 function randomBoundary(): string {
   let boundary = "--------------------------";
@@ -23,18 +25,31 @@ function randomBoundary(): string {
   return boundary;
 }
 
+/**
+ * Checks whether `buf` should be considered to match the boundary.
+ *
+ * The prefix is "--boundary" or "\r\n--boundary" or "\n--boundary", and the
+ * caller has verified already that `hasPrefix(buf, prefix)` is true.
+ *
+ * `matchAfterPrefix()` returns `1` if the buffer does match the boundary,
+ * meaning the prefix is followed by a dash, space, tab, cr, nl, or EOF.
+ *
+ * It returns `-1` if the buffer definitely does NOT match the boundary,
+ * meaning the prefix is followed by some other character.
+ * For example, "--foobar" does not match "--foo".
+ *
+ * It returns `0` more input needs to be read to make the decision,
+ * meaning that `buf.length` and `prefix.length` are the same.
+ */
 export function matchAfterPrefix(
-  a: Uint8Array,
+  buf: Uint8Array,
   prefix: Uint8Array,
-  bufState: BufState
+  eof: boolean
-): number {
+): -1 | 0 | 1 {
-  if (a.length === prefix.length) {
+  if (buf.length === prefix.length) {
-    if (bufState) {
+    return eof ? 1 : 0;
-      return 1;
-    }
-    return 0;
   }
-  const c = a[prefix.length];
+  const c = buf[prefix.length];
   if (
     c === " ".charCodeAt(0) ||
     c === "\t".charCodeAt(0) ||
@@ -47,105 +62,117 @@ export function matchAfterPrefix(
   return -1;
 }
 
+/**
+ * Scans `buf` to identify how much of it can be safely returned as part of the
+ * `PartReader` body.
+ *
+ * @param buf - The buffer to search for boundaries.
+ * @param dashBoundary - Is "--boundary".
+ * @param newLineDashBoundary - Is "\r\n--boundary" or "\n--boundary", depending
+ * on what mode we are in. The comments below (and the name) assume
+ * "\n--boundary", but either is accepted.
+ * @param total - The number of bytes read out so far. If total == 0, then a
+ * leading "--boundary" is recognized.
+ * @param eof - Whether `buf` contains the final bytes in the stream before EOF.
+ * If `eof` is false, more bytes are expected to follow.
+ * @returns The number of data bytes from buf that can be returned as part of
+ * the `PartReader` body.
+ */
 export function scanUntilBoundary(
   buf: Uint8Array,
   dashBoundary: Uint8Array,
   newLineDashBoundary: Uint8Array,
   total: number,
-  state: BufState
+  eof: boolean
-): [number, BufState] {
+): number | EOF {
   if (total === 0) {
-    if (bytes.hasPrefix(buf, dashBoundary)) {
+    // At beginning of body, allow dashBoundary.
-      switch (matchAfterPrefix(buf, dashBoundary, state)) {
+    if (hasPrefix(buf, dashBoundary)) {
+      switch (matchAfterPrefix(buf, dashBoundary, eof)) {
         case -1:
-          return [dashBoundary.length, null];
+          return dashBoundary.length;
         case 0:
-          return [0, null];
+          return 0;
         case 1:
-          return [0, "EOF"];
+          return EOF;
-      }
-      if (bytes.hasPrefix(dashBoundary, buf)) {
-        return [0, state];
       }
     }
+    if (hasPrefix(dashBoundary, buf)) {
+      return 0;
+    }
   }
-  const i = bytes.findIndex(buf, newLineDashBoundary);
+
+  // Search for "\n--boundary".
+  const i = findIndex(buf, newLineDashBoundary);
   if (i >= 0) {
-    switch (matchAfterPrefix(buf.slice(i), newLineDashBoundary, state)) {
+    switch (matchAfterPrefix(buf.slice(i), newLineDashBoundary, eof)) {
       case -1:
-        // eslint-disable-next-line @typescript-eslint/restrict-plus-operands
+        return i + newLineDashBoundary.length;
-        return [i + newLineDashBoundary.length, null];
       case 0:
-        return [i, null];
+        return i;
       case 1:
-        return [i, "EOF"];
+        return i > 0 ? i : EOF;
     }
   }
-  if (bytes.hasPrefix(newLineDashBoundary, buf)) {
+  if (hasPrefix(newLineDashBoundary, buf)) {
-    return [0, state];
+    return 0;
   }
-  const j = bytes.findLastIndex(buf, newLineDashBoundary.slice(0, 1));
+
-  if (j >= 0 && bytes.hasPrefix(newLineDashBoundary, buf.slice(j))) {
+  // Otherwise, anything up to the final \n is not part of the boundary and so
-    return [j, null];
+  // must be part of the body. Also, if the section from the final \n onward is
+  // not a prefix of the boundary, it too must be part of the body.
+  const j = findLastIndex(buf, newLineDashBoundary.slice(0, 1));
+  if (j >= 0 && hasPrefix(newLineDashBoundary, buf.slice(j))) {
+    return j;
   }
-  return [buf.length, state];
+
+  return buf.length;
 }
 
-let i = 0;
-
 class PartReader implements Reader, Closer {
-  n: number = 0;
+  n: number | EOF = 0;
   total: number = 0;
-  bufState: BufState = null;
-  index = i++;
 
   constructor(private mr: MultipartReader, public readonly headers: Headers) {}
 
   async read(p: Uint8Array): Promise<ReadResult> {
     const br = this.mr.bufReader;
-    const returnResult = (nread: number, bufState: BufState): ReadResult => {
+
-      if (bufState && bufState !== "EOF") {
+    // Read into buffer until we identify some data to return,
-        throw bufState;
+    // or we find a reason to stop (boundary or EOF).
+    let peekLength = 1;
+    while (this.n === 0) {
+      peekLength = max(peekLength, br.buffered());
+      const peekBuf = await br.peek(peekLength);
+      if (peekBuf === EOF) {
+        throw new UnexpectedEOFError();
       }
-      return { nread, eof: bufState === "EOF" };
+      const eof = peekBuf.length < peekLength;
-    };
+      this.n = scanUntilBoundary(
-    if (this.n === 0 && !this.bufState) {
+        peekBuf,
-      const [peek] = await br.peek(br.buffered());
-      const [n, state] = scanUntilBoundary(
-        peek,
         this.mr.dashBoundary,
         this.mr.newLineDashBoundary,
         this.total,
-        this.bufState
+        eof
       );
-      this.n = n;
+      if (this.n === 0) {
-      this.bufState = state;
+        // Force buffered I/O to read more into buffer.
-      if (this.n === 0 && !this.bufState) {
+        assertStrictEq(eof, false);
-        // eslint-disable-next-line @typescript-eslint/restrict-plus-operands
+        peekLength++;
-        const [, state] = await br.peek(peek.length + 1);
-        this.bufState = state;
-        if (this.bufState === "EOF") {
-          this.bufState = new RangeError("unexpected eof");
-        }
       }
     }
-    if (this.n === 0) {
+
-      return returnResult(0, this.bufState);
+    if (this.n === EOF) {
+      return { nread: 0, eof: true };
     }
 
-    let n = 0;
+    const nread = min(p.length, this.n);
-    if (p.byteLength > this.n) {
+    const buf = p.subarray(0, nread);
-      n = this.n;
+    const r = await br.readFull(buf);
-    }
+    assertStrictEq(r, buf);
-    const buf = p.slice(0, n);
-    const [nread] = await this.mr.bufReader.readFull(buf);
-    p.set(buf);
-    this.total += nread;
     this.n -= nread;
-    if (this.n === 0) {
+    this.total += nread;
-      return returnResult(n, this.bufState);
+    return { nread, eof: false };
-    }
-    return returnResult(n, null);
   }
 
   close(): void {}
@@ -212,7 +239,7 @@ export class MultipartReader {
   readonly dashBoundary = encoder.encode(`--${this.boundary}`);
   readonly bufReader: BufReader;
 
-  constructor(private reader: Reader, private boundary: string) {
+  constructor(reader: Reader, private boundary: string) {
     this.bufReader = new BufReader(reader);
   }
 
@@ -228,7 +255,7 @@ export class MultipartReader {
     const buf = new Buffer(new Uint8Array(maxValueBytes));
     for (;;) {
       const p = await this.nextPart();
-      if (!p) {
+      if (p === EOF) {
         break;
       }
       if (p.formName === "") {
@@ -251,7 +278,7 @@ export class MultipartReader {
       const n = await copy(buf, p);
       if (n > maxMemory) {
         // too big, write to disk and flush buffer
-        const ext = path.extname(p.fileName);
+        const ext = extname(p.fileName);
         const { file, filepath } = await tempFile(".", {
           prefix: "multipart-",
           postfix: ext
@@ -277,7 +304,7 @@ export class MultipartReader {
           filename: p.fileName,
           type: p.headers.get("content-type"),
           content: buf.bytes(),
-          size: buf.bytes().byteLength
+          size: buf.length
         };
         maxMemory -= n;
         maxValueBytes -= n;
@@ -290,35 +317,32 @@ export class MultipartReader {
   private currentPart: PartReader;
   private partsRead: number;
 
-  private async nextPart(): Promise<PartReader> {
+  private async nextPart(): Promise<PartReader | EOF> {
     if (this.currentPart) {
       this.currentPart.close();
     }
-    if (bytes.equal(this.dashBoundary, encoder.encode("--"))) {
+    if (equal(this.dashBoundary, encoder.encode("--"))) {
       throw new Error("boundary is empty");
     }
     let expectNewPart = false;
     for (;;) {
-      const [line, state] = await this.bufReader.readSlice("\n".charCodeAt(0));
+      const line = await this.bufReader.readSlice("\n".charCodeAt(0));
-      if (state === "EOF" && this.isFinalBoundary(line)) {
+      if (line === EOF) {
-        break;
+        throw new UnexpectedEOFError();
-      }
-      if (state) {
-        throw new Error(`aa${state.toString()}`);
       }
       if (this.isBoundaryDelimiterLine(line)) {
         this.partsRead++;
         const r = new TextProtoReader(this.bufReader);
-        const [headers, state] = await r.readMIMEHeader();
+        const headers = await r.readMIMEHeader();
-        if (state) {
+        if (headers === EOF) {
-          throw state;
+          throw new UnexpectedEOFError();
         }
         const np = new PartReader(this, headers);
         this.currentPart = np;
         return np;
       }
       if (this.isFinalBoundary(line)) {
-        break;
+        return EOF;
       }
       if (expectNewPart) {
         throw new Error(`expecting a new Part; got line ${line}`);
@@ -326,28 +350,28 @@ export class MultipartReader {
       if (this.partsRead === 0) {
         continue;
       }
-      if (bytes.equal(line, this.newLine)) {
+      if (equal(line, this.newLine)) {
         expectNewPart = true;
         continue;
       }
-      throw new Error(`unexpected line in next(): ${line}`);
+      throw new Error(`unexpected line in nextPart(): ${line}`);
     }
   }
 
   private isFinalBoundary(line: Uint8Array): boolean {
-    if (!bytes.hasPrefix(line, this.dashBoundaryDash)) {
+    if (!hasPrefix(line, this.dashBoundaryDash)) {
       return false;
     }
     let rest = line.slice(this.dashBoundaryDash.length, line.length);
-    return rest.length === 0 || bytes.equal(skipLWSPChar(rest), this.newLine);
+    return rest.length === 0 || equal(skipLWSPChar(rest), this.newLine);
   }
 
   private isBoundaryDelimiterLine(line: Uint8Array): boolean {
-    if (!bytes.hasPrefix(line, this.dashBoundary)) {
+    if (!hasPrefix(line, this.dashBoundary)) {
       return false;
     }
     const rest = line.slice(this.dashBoundary.length);
-    return bytes.equal(skipLWSPChar(rest), this.newLine);
+    return equal(skipLWSPChar(rest), this.newLine);
   }
 }
 
@@ -478,7 +502,7 @@ export class MultipartWriter {
     await copy(f, file);
   }
 
-  private flush(): Promise<BufState> {
+  private flush(): Promise<void> {
     return this.bufWriter.flush();
   }
 

mime/multipart_test.ts

@@ -7,7 +7,7 @@ import {
   assertThrows,
   assertThrowsAsync
 } from "../testing/asserts.ts";
-import { test } from "../testing/mod.ts";
+import { test, runIfMain } from "../testing/mod.ts";
 import {
   matchAfterPrefix,
   MultipartReader,
@@ -16,6 +16,7 @@ import {
 } from "./multipart.ts";
 import * as path from "../fs/path.ts";
 import { FormFile, isFormFile } from "../multipart/formfile.ts";
+import { EOF } from "../io/bufio.ts";
 import { StringWriter } from "../io/writers.ts";
 
 const e = new TextEncoder();
@@ -25,71 +26,67 @@ const nlDashBoundary = e.encode("\r\n--" + boundary);
 
 test(function multipartScanUntilBoundary1(): void {
   const data = `--${boundary}`;
-  const [n, err] = scanUntilBoundary(
+  const n = scanUntilBoundary(
     e.encode(data),
     dashBoundary,
     nlDashBoundary,
     0,
-    "EOF"
+    true
   );
-  assertEquals(n, 0);
+  assertEquals(n, EOF);
-  assertEquals(err, "EOF");
 });
 
 test(function multipartScanUntilBoundary2(): void {
   const data = `foo\r\n--${boundary}`;
-  const [n, err] = scanUntilBoundary(
+  const n = scanUntilBoundary(
     e.encode(data),
     dashBoundary,
     nlDashBoundary,
     0,
-    "EOF"
+    true
   );
   assertEquals(n, 3);
-  assertEquals(err, "EOF");
-});
-
-test(function multipartScanUntilBoundary4(): void {
-  const data = `foo\r\n--`;
-  const [n, err] = scanUntilBoundary(
-    e.encode(data),
-    dashBoundary,
-    nlDashBoundary,
-    0,
-    null
-  );
-  assertEquals(n, 3);
-  assertEquals(err, null);
 });
 
 test(function multipartScanUntilBoundary3(): void {
   const data = `foobar`;
-  const [n, err] = scanUntilBoundary(
+  const n = scanUntilBoundary(
     e.encode(data),
     dashBoundary,
     nlDashBoundary,
     0,
-    null
+    false
   );
   assertEquals(n, data.length);
-  assertEquals(err, null);
+});
+
+test(function multipartScanUntilBoundary4(): void {
+  const data = `foo\r\n--`;
+  const n = scanUntilBoundary(
+    e.encode(data),
+    dashBoundary,
+    nlDashBoundary,
+    0,
+    false
+  );
+  assertEquals(n, 3);
 });
 
 test(function multipartMatchAfterPrefix1(): void {
   const data = `${boundary}\r`;
-  const v = matchAfterPrefix(e.encode(data), e.encode(boundary), null);
+  const v = matchAfterPrefix(e.encode(data), e.encode(boundary), false);
   assertEquals(v, 1);
 });
 
 test(function multipartMatchAfterPrefix2(): void {
   const data = `${boundary}hoge`;
-  const v = matchAfterPrefix(e.encode(data), e.encode(boundary), null);
+  const v = matchAfterPrefix(e.encode(data), e.encode(boundary), false);
   assertEquals(v, -1);
 });
 
 test(function multipartMatchAfterPrefix3(): void {
   const data = `${boundary}`;
-  const v = matchAfterPrefix(e.encode(data), e.encode(boundary), null);
+  const v = matchAfterPrefix(e.encode(data), e.encode(boundary), false);
   assertEquals(v, 0);
 });
 
@@ -211,3 +208,5 @@ test(async function multipartMultipartReader2(): Promise<void> {
     await remove(file.tempfile);
   }
 });
+
+runIfMain(import.meta);

textproto/mod.ts

@@ -3,7 +3,7 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-import { BufReader, BufState } from "../io/bufio.ts";
+import { BufReader, EOF, UnexpectedEOFError } from "../io/bufio.ts";
 import { charCode } from "../io/util.ts";
 
 const asciiDecoder = new TextDecoder();
@@ -39,9 +39,10 @@ export class TextProtoReader {
   /** readLine() reads a single line from the TextProtoReader,
    * eliding the final \n or \r\n from the returned string.
    */
-  async readLine(): Promise<[string, BufState]> {
+  async readLine(): Promise<string | EOF> {
-    let [line, err] = await this.readLineSlice();
+    const s = await this.readLineSlice();
-    return [str(line), err];
+    if (s === EOF) return EOF;
+    return str(s);
   }
 
   /** ReadMIMEHeader reads a MIME-style header from r.
@@ -64,29 +65,31 @@ export class TextProtoReader {
    *		"Long-Key": {"Even Longer Value"},
    *	}
    */
-  async readMIMEHeader(): Promise<[Headers, BufState]> {
+  async readMIMEHeader(): Promise<Headers | EOF> {
     let m = new Headers();
     let line: Uint8Array;
 
     // The first line cannot start with a leading space.
-    let [buf, err] = await this.r.peek(1);
+    let buf = await this.r.peek(1);
-    if (buf[0] == charCode(" ") || buf[0] == charCode("\t")) {
+    if (buf === EOF) {
-      [line, err] = await this.readLineSlice();
+      return EOF;
+    } else if (buf[0] == charCode(" ") || buf[0] == charCode("\t")) {
+      line = (await this.readLineSlice()) as Uint8Array;
     }
 
-    [buf, err] = await this.r.peek(1);
+    buf = await this.r.peek(1);
-    if (err == null && (buf[0] == charCode(" ") || buf[0] == charCode("\t"))) {
+    if (buf === EOF) {
+      throw new UnexpectedEOFError();
+    } else if (buf[0] == charCode(" ") || buf[0] == charCode("\t")) {
       throw new ProtocolError(
         `malformed MIME header initial line: ${str(line)}`
       );
     }
 
     while (true) {
-      let [kv, err] = await this.readLineSlice(); // readContinuedLineSlice
+      let kv = await this.readLineSlice(); // readContinuedLineSlice
-
+      if (kv === EOF) throw new UnexpectedEOFError();
-      if (kv.byteLength === 0) {
+      if (kv.byteLength === 0) return m;
-        return [m, err];
-      }
 
       // Key ends at first colon; should not have trailing spaces
       // but they appear in the wild, violating specs, so we remove
@@ -125,29 +128,26 @@ export class TextProtoReader {
       try {
         m.append(key, value);
       } catch {}
-
-      if (err != null) {
-        throw err;
-      }
     }
   }
 
-  async readLineSlice(): Promise<[Uint8Array, BufState]> {
+  async readLineSlice(): Promise<Uint8Array | EOF> {
     // this.closeDot();
     let line: Uint8Array;
     while (true) {
-      let [l, more, err] = await this.r.readLine();
+      const r = await this.r.readLine();
-      if (err != null) {
+      if (r === EOF) return EOF;
-        // Go's len(typed nil) works fine, but not in JS
+      const { line: l, more } = r;
-        return [new Uint8Array(0), err];
-      }
 
       // Avoid the copy if the first call produced a full line.
-      if (line == null && !more) {
+      if (!line && !more) {
+        // TODO(ry):
+        // This skipSpace() is definitely misplaced, but I don't know where it
+        // comes from nor how to fix it.
         if (this.skipSpace(l) === 0) {
-          return [new Uint8Array(0), null];
+          return new Uint8Array(0);
         }
-        return [l, null];
+        return l;
       }
 
       line = append(line, l);
@@ -155,7 +155,7 @@ export class TextProtoReader {
         break;
       }
     }
-    return [line, null];
+    return line;
   }
 
   skipSpace(l: Uint8Array): number {

textproto/reader_test.ts

@@ -3,11 +3,21 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-import { BufReader } from "../io/bufio.ts";
+import { BufReader, EOF } from "../io/bufio.ts";
 import { TextProtoReader, ProtocolError } from "./mod.ts";
 import { stringsReader } from "../io/util.ts";
-import { assert, assertEquals, assertThrows } from "../testing/asserts.ts";
+import {
-import { test } from "../testing/mod.ts";
+  assert,
+  assertEquals,
+  assertNotEquals,
+  assertThrows
+} from "../testing/asserts.ts";
+import { test, runIfMain } from "../testing/mod.ts";
+
+function assertNotEOF<T extends {}>(val: T | EOF): T {
+  assertNotEquals(val, EOF);
+  return val as T;
+}
 
 function reader(s: string): TextProtoReader {
   return new TextProtoReader(new BufReader(stringsReader(s)));
@@ -21,25 +31,21 @@ function reader(s: string): TextProtoReader {
 // });
 
 test(async function textprotoReadEmpty(): Promise<void> {
-  let r = reader("");
+  const r = reader("");
-  let [, err] = await r.readMIMEHeader();
+  const m = await r.readMIMEHeader();
-  // Should not crash!
+  assertEquals(m, EOF);
-  assertEquals(err, "EOF");
 });
 
 test(async function textprotoReader(): Promise<void> {
-  let r = reader("line1\nline2\n");
+  const r = reader("line1\nline2\n");
-  let [s, err] = await r.readLine();
+  let s = await r.readLine();
   assertEquals(s, "line1");
-  assert(err == null);
 
-  [s, err] = await r.readLine();
+  s = await r.readLine();
   assertEquals(s, "line2");
-  assert(err == null);
 
-  [s, err] = await r.readLine();
+  s = await r.readLine();
-  assertEquals(s, "");
+  assert(s === EOF);
-  assert(err == "EOF");
 });
 
 test({
@@ -48,10 +54,9 @@ test({
     const input =
       "my-key: Value 1  \r\nLong-key: Even Longer Value\r\nmy-Key: Value 2\r\n\n";
     const r = reader(input);
-    const [m, err] = await r.readMIMEHeader();
+    const m = assertNotEOF(await r.readMIMEHeader());
     assertEquals(m.get("My-Key"), "Value 1, Value 2");
     assertEquals(m.get("Long-key"), "Even Longer Value");
-    assert(!err);
   }
 });
 
@@ -60,9 +65,8 @@ test({
   async fn(): Promise<void> {
     const input = "Foo: bar\n\n";
     const r = reader(input);
-    let [m, err] = await r.readMIMEHeader();
+    const m = assertNotEOF(await r.readMIMEHeader());
     assertEquals(m.get("Foo"), "bar");
-    assert(!err);
   }
 });
 
@@ -71,9 +75,8 @@ test({
   async fn(): Promise<void> {
     const input = ": bar\ntest-1: 1\n\n";
     const r = reader(input);
-    let [m, err] = await r.readMIMEHeader();
+    const m = assertNotEOF(await r.readMIMEHeader());
     assertEquals(m.get("Test-1"), "1");
-    assert(!err);
   }
 });
 
@@ -86,11 +89,9 @@ test({
       data.push("x");
     }
     const sdata = data.join("");
-    const r = reader(`Cookie: ${sdata}\r\n`);
+    const r = reader(`Cookie: ${sdata}\r\n\r\n`);
-    let [m] = await r.readMIMEHeader();
+    const m = assertNotEOF(await r.readMIMEHeader());
     assertEquals(m.get("Cookie"), sdata);
-    // TODO re-enable, here err === "EOF" is has to be null
-    // assert(!err);
   }
 });
 
@@ -106,12 +107,11 @@ test({
       "Audio Mode : None\r\n" +
       "Privilege : 127\r\n\r\n";
     const r = reader(input);
-    let [m, err] = await r.readMIMEHeader();
+    const m = assertNotEOF(await r.readMIMEHeader());
     assertEquals(m.get("Foo"), "bar");
     assertEquals(m.get("Content-Language"), "en");
     assertEquals(m.get("SID"), "0");
     assertEquals(m.get("Privilege"), "127");
-    assert(!err);
     // Not a legal http header
     assertThrows(
       (): void => {
@@ -176,9 +176,10 @@ test({
       "------WebKitFormBoundaryimeZ2Le9LjohiUiG--\r\n\n"
     ];
     const r = reader(input.join(""));
-    let [m, err] = await r.readMIMEHeader();
+    const m = assertNotEOF(await r.readMIMEHeader());
     assertEquals(m.get("Accept"), "*/*");
     assertEquals(m.get("Content-Disposition"), 'form-data; name="test"');
-    assert(!err);
   }
 });
+
+runIfMain(import.meta);

ws/mod.ts

@@ -4,7 +4,7 @@ import { decode, encode } from "../strings/mod.ts";
 
 type Conn = Deno.Conn;
 type Writer = Deno.Writer;
-import { BufReader, BufWriter } from "../io/bufio.ts";
+import { BufReader, BufWriter, EOF, UnexpectedEOFError } from "../io/bufio.ts";
 import { readLong, readShort, sliceLongToBytes } from "../io/ioutil.ts";
 import { Sha1 } from "./sha1.ts";
 import { writeResponse } from "../http/server.ts";
@@ -130,8 +130,7 @@ export async function writeFrame(
   header = append(header, frame.payload);
   const w = BufWriter.create(writer);
   await w.write(header);
-  const err = await w.flush();
+  await w.flush();
-  if (err) throw err;
 }
 
 /** Read websocket frame from given BufReader */
@@ -403,15 +402,71 @@ export function createSecKey(): string {
   return btoa(key);
 }
 
+async function handshake(
+  url: URL,
+  headers: Headers,
+  bufReader: BufReader,
+  bufWriter: BufWriter
+): Promise<void> {
+  const { hostname, pathname, searchParams } = url;
+  const key = createSecKey();
+
+  if (!headers.has("host")) {
+    headers.set("host", hostname);
+  }
+  headers.set("upgrade", "websocket");
+  headers.set("connection", "upgrade");
+  headers.set("sec-websocket-key", key);
+
+  let headerStr = `GET ${pathname}?${searchParams || ""} HTTP/1.1\r\n`;
+  for (const [key, value] of headers) {
+    headerStr += `${key}: ${value}\r\n`;
+  }
+  headerStr += "\r\n";
+
+  await bufWriter.write(encode(headerStr));
+  await bufWriter.flush();
+
+  const tpReader = new TextProtoReader(bufReader);
+  const statusLine = await tpReader.readLine();
+  if (statusLine === EOF) {
+    throw new UnexpectedEOFError();
+  }
+  const m = statusLine.match(/^(?<version>\S+) (?<statusCode>\S+) /);
+  if (!m) {
+    throw new Error("ws: invalid status line: " + statusLine);
+  }
+  const { version, statusCode } = m.groups;
+  if (version !== "HTTP/1.1" || statusCode !== "101") {
+    throw new Error(
+      `ws: server didn't accept handshake: ` +
+        `version=${version}, statusCode=${statusCode}`
+    );
+  }
+
+  const responseHeaders = await tpReader.readMIMEHeader();
+  if (responseHeaders === EOF) {
+    throw new UnexpectedEOFError();
+  }
+
+  const expectedSecAccept = createSecAccept(key);
+  const secAccept = responseHeaders.get("sec-websocket-accept");
+  if (secAccept !== expectedSecAccept) {
+    throw new Error(
+      `ws: unexpected sec-websocket-accept header: ` +
+        `expected=${expectedSecAccept}, actual=${secAccept}`
+    );
+  }
+}
+
 /** Connect to given websocket endpoint url. Endpoint must be acceptable for URL */
 export async function connectWebSocket(
   endpoint: string,
   headers: Headers = new Headers()
 ): Promise<WebSocket> {
   const url = new URL(endpoint);
-  const { hostname, pathname, searchParams } = url;
+  let { hostname, port } = url;
-  let port = url.port;
+  if (!port) {
-  if (!url.port) {
     if (url.protocol === "http" || url.protocol === "ws") {
       port = "80";
     } else if (url.protocol === "https" || url.protocol === "wss") {
@@ -419,62 +474,13 @@ export async function connectWebSocket(
     }
   }
   const conn = await Deno.dial("tcp", `${hostname}:${port}`);
-  const abortHandshake = (err: Error): void => {
-    conn.close();
-    throw err;
-  };
   const bufWriter = new BufWriter(conn);
   const bufReader = new BufReader(conn);
-  await bufWriter.write(
+  try {
-    encode(`GET ${pathname}?${searchParams || ""} HTTP/1.1\r\n`)
+    await handshake(url, headers, bufReader, bufWriter);
-  );
+  } catch (err) {
-  const key = createSecKey();
+    conn.close();
-  if (!headers.has("host")) {
+    throw err;
-    headers.set("host", hostname);
-  }
-  headers.set("upgrade", "websocket");
-  headers.set("connection", "upgrade");
-  headers.set("sec-websocket-key", key);
-  let headerStr = "";
-  for (const [key, value] of headers) {
-    headerStr += `${key}: ${value}\r\n`;
-  }
-  headerStr += "\r\n";
-  await bufWriter.write(encode(headerStr));
-  let err, statusLine, responseHeaders;
-  err = await bufWriter.flush();
-  if (err) {
-    throw new Error("ws: failed to send handshake: " + err);
-  }
-  const tpReader = new TextProtoReader(bufReader);
-  [statusLine, err] = await tpReader.readLine();
-  if (err) {
-    abortHandshake(new Error("ws: failed to read status line: " + err));
-  }
-  const m = statusLine.match(/^(.+?) (.+?) (.+?)$/);
-  if (!m) {
-    abortHandshake(new Error("ws: invalid status line: " + statusLine));
-  }
-  const [_, version, statusCode] = m;
-  if (version !== "HTTP/1.1" || statusCode !== "101") {
-    abortHandshake(
-      new Error(
-        `ws: server didn't accept handshake: version=${version}, statusCode=${statusCode}`
-      )
-    );
-  }
-  [responseHeaders, err] = await tpReader.readMIMEHeader();
-  if (err) {
-    abortHandshake(new Error("ws: failed to parse response headers: " + err));
-  }
-  const expectedSecAccept = createSecAccept(key);
-  const secAccept = responseHeaders.get("sec-websocket-accept");
-  if (secAccept !== expectedSecAccept) {
-    abortHandshake(
-      new Error(
-        `ws: unexpected sec-websocket-accept header: expected=${expectedSecAccept}, actual=${secAccept}`
-      )
-    );
   }
   return new WebSocketImpl(conn, {
     bufWriter,

ws/test.ts

@@ -107,8 +107,9 @@ test(async function wsReadUnmaskedPingPongFrame(): Promise<void> {
 });
 
 test(async function wsReadUnmaskedBigBinaryFrame(): Promise<void> {
+  const payloadLength = 0x100;
   const a = [0x82, 0x7e, 0x01, 0x00];
-  for (let i = 0; i < 256; i++) {
+  for (let i = 0; i < payloadLength; i++) {
     a.push(i);
   }
   const buf = new BufReader(new Buffer(new Uint8Array(a)));
@@ -116,12 +117,13 @@ test(async function wsReadUnmaskedBigBinaryFrame(): Promise<void> {
   assertEquals(bin.opcode, OpCode.BinaryFrame);
   assertEquals(bin.isLastFrame, true);
   assertEquals(bin.mask, undefined);
-  assertEquals(bin.payload.length, 256);
+  assertEquals(bin.payload.length, payloadLength);
 });
 
 test(async function wsReadUnmaskedBigBigBinaryFrame(): Promise<void> {
+  const payloadLength = 0x10000;
   const a = [0x82, 0x7f, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00];
-  for (let i = 0; i < 0xffff; i++) {
+  for (let i = 0; i < payloadLength; i++) {
     a.push(i);
   }
   const buf = new BufReader(new Buffer(new Uint8Array(a)));
@@ -129,7 +131,7 @@ test(async function wsReadUnmaskedBigBigBinaryFrame(): Promise<void> {
   assertEquals(bin.opcode, OpCode.BinaryFrame);
   assertEquals(bin.isLastFrame, true);
   assertEquals(bin.mask, undefined);
-  assertEquals(bin.payload.length, 0xffff + 1);
+  assertEquals(bin.payload.length, payloadLength);
 });
 
 test(async function wsCreateSecAccept(): Promise<void> {