Merge branch 'master' into ws_acceptable

fs/ensure_symlink.ts

@@ -4,8 +4,6 @@ import { ensureDir, ensureDirSync } from "./ensure_dir.ts";
 import { exists, existsSync } from "./exists.ts";
 import { getFileInfoType } from "./utils.ts";
 
-const isWindows = Deno.platform.os === "win";
-
 /**
  * Ensures that the link exists.
  * If the directory structure does not exist, it is created.
@@ -30,12 +28,7 @@ export async function ensureSymlink(src: string, dest: string): Promise<void> {
 
   await ensureDir(path.dirname(dest));
 
-  // TODO(axetroy): remove this if condition. refs: https://github.com/denoland/deno/issues/2169
-  if (isWindows) {
-    await Deno.symlink(src, dest, srcFilePathType || undefined);
-  } else {
-    await Deno.symlink(src, dest);
-  }
+  await Deno.symlink(src, dest, srcFilePathType);
 }
 
 /**
@@ -62,10 +55,5 @@ export function ensureSymlinkSync(src: string, dest: string): void {
 
   ensureDirSync(path.dirname(dest));
 
-  // TODO(axetroy): remove this if condition. refs: https://github.com/denoland/deno/issues/2169
-  if (isWindows) {
-    Deno.symlinkSync(src, dest, srcFilePathType || undefined);
-  } else {
-    Deno.symlinkSync(src, dest);
-  }
+  Deno.symlinkSync(src, dest, srcFilePathType);
 }

io/bufio.ts

@@ -33,6 +33,11 @@ export class BufReader implements Reader {
   private lastCharSize: number;
   private err: BufState;
 
+  /** return new BufReader unless r is BufReader */
+  static create(r: Reader, size = DEFAULT_BUF_SIZE): BufReader {
+    return r instanceof BufReader ? r : new BufReader(r, size);
+  }
+
   constructor(rd: Reader, size = DEFAULT_BUF_SIZE) {
     if (size < MIN_BUF_SIZE) {
       size = MIN_BUF_SIZE;
@@ -368,6 +373,11 @@ export class BufWriter implements Writer {
   n: number = 0;
   err: null | BufState = null;
 
+  /** return new BufWriter unless w is BufWriter */
+  static create(w: Writer, size = DEFAULT_BUF_SIZE): BufWriter {
+    return w instanceof BufWriter ? w : new BufWriter(w, size);
+  }
+
   constructor(private wr: Writer, size = DEFAULT_BUF_SIZE) {
     if (size <= 0) {
       size = DEFAULT_BUF_SIZE;

ws/example_client.ts

@@ -0,0 +1,55 @@
+import {
+  connectWebSocket,
+  isWebSocketCloseEvent,
+  isWebSocketPingEvent,
+  isWebSocketPongEvent
+} from "../ws/mod.ts";
+import { encode } from "../strings/strings.ts";
+import { BufReader } from "../io/bufio.ts";
+import { TextProtoReader } from "../textproto/mod.ts";
+import { blue, green, red, yellow } from "../colors/mod.ts";
+
+const endpoint = Deno.args[1] || "ws://127.0.0.1:8080";
+/** simple websocket cli */
+async function main(): Promise<void> {
+  const sock = await connectWebSocket(endpoint);
+  console.log(green("ws connected! (type 'close' to quit)"));
+  (async function(): Promise<void> {
+    for await (const msg of sock.receive()) {
+      if (typeof msg === "string") {
+        console.log(yellow("< " + msg));
+      } else if (isWebSocketPingEvent(msg)) {
+        console.log(blue("< ping"));
+      } else if (isWebSocketPongEvent(msg)) {
+        console.log(blue("< pong"));
+      } else if (isWebSocketCloseEvent(msg)) {
+        console.log(red(`closed: code=${msg.code}, reason=${msg.reason}`));
+      }
+    }
+  })();
+  const tpr = new TextProtoReader(new BufReader(Deno.stdin));
+  while (true) {
+    await Deno.stdout.write(encode("> "));
+    const [line, err] = await tpr.readLine();
+    if (err) {
+      console.error(red(`failed to read line from stdin: ${err}`));
+      break;
+    }
+    if (line === "close") {
+      break;
+    } else if (line === "ping") {
+      await sock.ping();
+    } else {
+      await sock.send(line);
+    }
+    // FIXME: Without this, sock.receive() won't resolved though it is readable...
+    await new Promise((resolve): void => setTimeout(resolve, 0));
+  }
+  await sock.close(1000);
+  // FIXME: conn.close() won't shutdown process...
+  Deno.exit(0);
+}
+
+if (import.meta.main) {
+  main();
+}

ws/example_server.ts

@@ -0,0 +1,66 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+import { serve } from "../http/server.ts";
+import {
+  acceptWebSocket,
+  isWebSocketCloseEvent,
+  isWebSocketPingEvent,
+  WebSocket
+} from "./mod.ts";
+
+/** websocket echo server */
+const port = Deno.args[1] || "8080";
+async function main(): Promise<void> {
+  console.log(`websocket server is running on :${port}`);
+  for await (const req of serve(`:${port}`)) {
+    const { headers, conn } = req;
+    acceptWebSocket({
+      conn,
+      headers,
+      bufReader: req.r,
+      bufWriter: req.w
+    })
+      .then(
+        async (sock: WebSocket): Promise<void> => {
+          console.log("socket connected!");
+          const it = sock.receive();
+          while (true) {
+            try {
+              const { done, value } = await it.next();
+              if (done) {
+                break;
+              }
+              const ev = value;
+              if (typeof ev === "string") {
+                // text message
+                console.log("ws:Text", ev);
+                await sock.send(ev);
+              } else if (ev instanceof Uint8Array) {
+                // binary message
+                console.log("ws:Binary", ev);
+              } else if (isWebSocketPingEvent(ev)) {
+                const [, body] = ev;
+                // ping
+                console.log("ws:Ping", body);
+              } else if (isWebSocketCloseEvent(ev)) {
+                // close
+                const { code, reason } = ev;
+                console.log("ws:Close", code, reason);
+              }
+            } catch (e) {
+              console.error(`failed to receive frame: ${e}`);
+              await sock.close(1000).catch(console.error);
+            }
+          }
+        }
+      )
+      .catch(
+        (err: Error): void => {
+          console.error(`failed to accept websocket: ${err}`);
+        }
+      );
+  }
+}
+
+if (import.meta.main) {
+  main();
+}