6 files changed, 149 insertions, 141 deletions

diff --git a/js/errors.ts b/js/errors.ts
index 11d4cd509..2d0572e6d 100644
--- a/js/errors.ts
+++ b/js/errors.ts
@@ -10,8 +10,17 @@ export class DenoError<T extends fbs.ErrorKind> extends Error {
 
 // @internal
 export function maybeThrowError(base: fbs.Base): void {
+  const err = maybeError(base);
+  if (err != null) {
+    throw err;
+  }
+}
+
+export function maybeError(base: fbs.Base): null | DenoError<fbs.ErrorKind> {
   const kind = base.errorKind();
-  if (kind !== fbs.ErrorKind.NoError) {
-    throw new DenoError(kind, base.error()!);
+  if (kind === fbs.ErrorKind.NoError) {
+    return null;
+  } else {
+    return new DenoError(kind, base.error()!);
   }
 }
diff --git a/js/fbs_util.ts b/js/fbs_util.ts
index bb623d54d..16f3b6ca2 100644
--- a/js/fbs_util.ts
+++ b/js/fbs_util.ts
@@ -1,27 +1,77 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
+// TODO Rename this file to //js/dispatch.ts
 import { libdeno } from "./libdeno";
 import { flatbuffers } from "flatbuffers";
-import { maybeThrowError } from "./errors";
 import { deno as fbs } from "gen/msg_generated";
+import * as errors from "./errors";
+import * as util from "./util";
+
+let nextCmdId = 0;
+const promiseTable = new Map<number, util.Resolvable<fbs.Base>>();
+
+export function handleAsyncMsgFromRust(ui8: Uint8Array) {
+  const bb = new flatbuffers.ByteBuffer(ui8);
+  const base = fbs.Base.getRootAsBase(bb);
+
+  const cmdId = base.cmdId();
+  const promise = promiseTable.get(cmdId);
+  util.assert(promise != null, `Expecting promise in table. ${cmdId}`);
+  promiseTable.delete(cmdId);
+  const err = errors.maybeError(base);
+  if (err != null) {
+    promise!.reject(err);
+  } else {
+    promise!.resolve(base);
+  }
+}
 
 // @internal
+export function sendAsync(
+  builder: flatbuffers.Builder,
+  msgType: fbs.Any,
+  msg: flatbuffers.Offset
+): Promise<fbs.Base> {
+  const [cmdId, resBuf] = sendInternal(builder, msgType, msg, false);
+  util.assert(resBuf == null);
+  const promise = util.createResolvable<fbs.Base>();
+  promiseTable.set(cmdId, promise);
+  return promise;
+}
+
+// TODO Rename to sendSync
+// @internal
 export function send(
   builder: flatbuffers.Builder,
   msgType: fbs.Any,
   msg: flatbuffers.Offset
 ): null | fbs.Base {
-  fbs.Base.startBase(builder);
-  fbs.Base.addMsg(builder, msg);
-  fbs.Base.addMsgType(builder, msgType);
-  builder.finish(fbs.Base.endBase(builder));
-
-  const resBuf = libdeno.send(builder.asUint8Array());
+  const [cmdId, resBuf] = sendInternal(builder, msgType, msg, true);
+  util.assert(cmdId >= 0);
   if (resBuf == null) {
     return null;
   } else {
-    const bb = new flatbuffers.ByteBuffer(new Uint8Array(resBuf!));
+    const u8 = new Uint8Array(resBuf!);
+    // console.log("recv sync message", util.hexdump(u8));
+    const bb = new flatbuffers.ByteBuffer(u8);
     const baseRes = fbs.Base.getRootAsBase(bb);
-    maybeThrowError(baseRes);
+    errors.maybeThrowError(baseRes);
     return baseRes;
   }
 }
+
+function sendInternal(
+  builder: flatbuffers.Builder,
+  msgType: fbs.Any,
+  msg: flatbuffers.Offset,
+  sync = true
+): [number, null | Uint8Array] {
+  const cmdId = nextCmdId++;
+  fbs.Base.startBase(builder);
+  fbs.Base.addMsg(builder, msg);
+  fbs.Base.addMsgType(builder, msgType);
+  fbs.Base.addSync(builder, sync);
+  fbs.Base.addCmdId(builder, cmdId);
+  builder.finish(fbs.Base.endBase(builder));
+
+  return [cmdId, libdeno.send(builder.asUint8Array())];
+}
diff --git a/js/fetch.ts b/js/fetch.ts
index 2d57fa65c..20af1d03d 100644
--- a/js/fetch.ts
+++ b/js/fetch.ts
@@ -8,7 +8,7 @@ import {
   notImplemented
 } from "./util";
 import { flatbuffers } from "flatbuffers";
-import { send } from "./fbs_util";
+import { sendAsync } from "./fbs_util";
 import { deno as fbs } from "gen/msg_generated";
 import {
   Headers,
@@ -20,16 +20,6 @@ import {
 } from "./fetch_types";
 import { TextDecoder } from "./text_encoding";
 
-/** @internal */
-export function onFetchRes(base: fbs.Base, msg: fbs.FetchRes) {
-  const id = msg.id();
-  const req = fetchRequests.get(id);
-  assert(req != null, `Couldn't find FetchRequest id ${id}`);
-  req!.onMsg(base, msg);
-}
-
-const fetchRequests = new Map<number, FetchRequest>();
-
 class DenoHeaders implements Headers {
   append(name: string, value: string): void {
     assert(false, "Implement me");
@@ -58,10 +48,9 @@ class DenoHeaders implements Headers {
 }
 
 class FetchResponse implements Response {
-  readonly url: string;
+  readonly url: string = "";
   body: null;
   bodyUsed = false; // TODO
-  status = 0;
   statusText = "FIXME"; // TODO
   readonly type = "basic"; // TODO
   redirected = false; // TODO
@@ -71,10 +60,12 @@ class FetchResponse implements Response {
   private first = true;
   private bodyWaiter: Resolvable<ArrayBuffer>;
 
-  constructor(readonly req: FetchRequest) {
-    this.url = req.url;
+  constructor(readonly status: number, readonly body_: ArrayBuffer) {
     this.bodyWaiter = createResolvable();
     this.trailer = createResolvable();
+    setTimeout(() => {
+      this.bodyWaiter.resolve(body_);
+    }, 0);
   }
 
   arrayBuffer(): Promise<ArrayBuffer> {
@@ -114,78 +105,50 @@ class FetchResponse implements Response {
   onHeader?: (res: FetchResponse) => void;
   onError?: (error: Error) => void;
 
-  onMsg(base: fbs.Base, msg: fbs.FetchRes) {
+  onMsg(base: fbs.Base) {
+    /*
     const error = base.error();
     if (error != null) {
       assert(this.onError != null);
       this.onError!(new Error(error));
       return;
     }
+    */
 
     if (this.first) {
       this.first = false;
-      this.status = msg.status();
-      assert(this.onHeader != null);
-      this.onHeader!(this);
-    } else {
-      // Body message. Assuming it all comes in one message now.
-      const bodyArray = msg.bodyArray();
-      assert(bodyArray != null);
-      const ab = typedArrayToArrayBuffer(bodyArray!);
-      this.bodyWaiter.resolve(ab);
     }
   }
 }
 
-let nextFetchId = 0;
-//TODO implements Request
-class FetchRequest {
-  private readonly id: number;
-  response: FetchResponse;
-  constructor(readonly url: string) {
-    this.id = nextFetchId++;
-    fetchRequests.set(this.id, this);
-    this.response = new FetchResponse(this);
-  }
-
-  onMsg(base: fbs.Base, msg: fbs.FetchRes) {
-    this.response.onMsg(base, msg);
-  }
-
-  destroy() {
-    fetchRequests.delete(this.id);
-  }
-
-  start() {
-    log("dispatch FETCH_REQ", this.id, this.url);
-
-    // Send FetchReq message
-    const builder = new flatbuffers.Builder();
-    const url = builder.createString(this.url);
-    fbs.FetchReq.startFetchReq(builder);
-    fbs.FetchReq.addId(builder, this.id);
-    fbs.FetchReq.addUrl(builder, url);
-    const msg = fbs.FetchReq.endFetchReq(builder);
-    send(builder, fbs.Any.FetchReq, msg);
-  }
-}
-
-export function fetch(
+export async function fetch(
   input?: Request | string,
   init?: RequestInit
 ): Promise<Response> {
-  const fetchReq = new FetchRequest(input as string);
-  const response = fetchReq.response;
-  const promise = new Promise<Response>((resolve, reject) => {
-    response.onHeader = (response: FetchResponse) => {
-      log("onHeader");
-      resolve(response);
-    };
-    response.onError = (error: Error) => {
-      log("onError", error);
-      reject(error);
-    };
-  });
-  fetchReq.start();
-  return promise;
+  const url = input as string;
+  log("dispatch FETCH_REQ", url);
+
+  // Send FetchReq message
+  const builder = new flatbuffers.Builder();
+  const url_ = builder.createString(url);
+  fbs.FetchReq.startFetchReq(builder);
+  fbs.FetchReq.addUrl(builder, url_);
+  const resBase = await sendAsync(
+    builder,
+    fbs.Any.FetchReq,
+    fbs.FetchReq.endFetchReq(builder)
+  );
+
+  // Decode FetchRes
+  assert(fbs.Any.FetchRes === resBase.msgType());
+  const msg = new fbs.FetchRes();
+  assert(resBase.msg(msg) != null);
+
+  const status = msg.status();
+  const bodyArray = msg.bodyArray();
+  assert(bodyArray != null);
+  const body = typedArrayToArrayBuffer(bodyArray!);
+
+  const response = new FetchResponse(status, body);
+  return response;
 }
diff --git a/js/main.ts b/js/main.ts
index 538d0cc27..e45562d99 100644
--- a/js/main.ts
+++ b/js/main.ts
@@ -5,10 +5,8 @@ import { assert, log, setLogDebug } from "./util";
 import * as os from "./os";
 import { DenoCompiler } from "./compiler";
 import { libdeno } from "./libdeno";
-import * as timers from "./timers";
-import { onFetchRes } from "./fetch";
 import { argv } from "./deno";
-import { send } from "./fbs_util";
+import { send, handleAsyncMsgFromRust } from "./fbs_util";
 
 function sendStart(): fbs.StartRes {
   const builder = new flatbuffers.Builder();
@@ -22,29 +20,6 @@ function sendStart(): fbs.StartRes {
   return startRes;
 }
 
-function onMessage(ui8: Uint8Array) {
-  const bb = new flatbuffers.ByteBuffer(ui8);
-  const base = fbs.Base.getRootAsBase(bb);
-  switch (base.msgType()) {
-    case fbs.Any.FetchRes: {
-      const msg = new fbs.FetchRes();
-      assert(base.msg(msg) != null);
-      onFetchRes(base, msg);
-      break;
-    }
-    case fbs.Any.TimerReady: {
-      const msg = new fbs.TimerReady();
-      assert(base.msg(msg) != null);
-      timers.onMessage(msg);
-      break;
-    }
-    default: {
-      assert(false, "Unhandled message type");
-      break;
-    }
-  }
-}
-
 function onGlobalError(
   message: string,
   source: string,
@@ -58,7 +33,7 @@ function onGlobalError(
 
 /* tslint:disable-next-line:no-default-export */
 export default function denoMain() {
-  libdeno.recv(onMessage);
+  libdeno.recv(handleAsyncMsgFromRust);
   libdeno.setGlobalErrorHandler(onGlobalError);
   const compiler = DenoCompiler.instance();
 
diff --git a/js/timers.ts b/js/timers.ts
index d09af0295..6b23c64f1 100644
--- a/js/timers.ts
+++ b/js/timers.ts
@@ -3,7 +3,7 @@ import { assert } from "./util";
 import * as util from "./util";
 import { deno as fbs } from "gen/msg_generated";
 import { flatbuffers } from "flatbuffers";
-import { send } from "./fbs_util";
+import { send, sendAsync } from "./fbs_util";
 
 let nextTimerId = 1;
 
@@ -19,50 +19,51 @@ interface Timer {
   delay: number; // milliseconds
 }
 
-const timers = new Map<number, Timer>();
-
-/** @internal */
-export function onMessage(msg: fbs.TimerReady) {
-  const timerReadyId = msg.id();
-  const timerReadyDone = msg.done();
-  const timer = timers.get(timerReadyId);
-  if (!timer) {
-    return;
-  }
-  timer.cb(...timer.args);
-  if (timerReadyDone) {
-    timers.delete(timerReadyId);
-  }
-}
-
 function startTimer(
+  id: number,
   cb: TimerCallback,
   delay: number,
   interval: boolean,
   // tslint:disable-next-line:no-any
   args: any[]
-): number {
-  const timer = {
-    id: nextTimerId++,
+): void {
+  const timer: Timer = {
+    id,
     interval,
     delay,
     args,
     cb
   };
-  timers.set(timer.id, timer);
-
   util.log("timers.ts startTimer");
 
   // Send TimerStart message
   const builder = new flatbuffers.Builder();
   fbs.TimerStart.startTimerStart(builder);
   fbs.TimerStart.addId(builder, timer.id);
-  fbs.TimerStart.addInterval(builder, timer.interval);
   fbs.TimerStart.addDelay(builder, timer.delay);
   const msg = fbs.TimerStart.endTimerStart(builder);
-  const baseRes = send(builder, fbs.Any.TimerStart, msg);
-  assert(baseRes == null);
-  return timer.id;
+
+  sendAsync(builder, fbs.Any.TimerStart, msg).then(
+    baseRes => {
+      assert(fbs.Any.TimerReady === baseRes!.msgType());
+      const msg = new fbs.TimerReady();
+      assert(baseRes!.msg(msg) != null);
+      assert(msg.id() === timer.id);
+      if (msg.canceled()) {
+        util.log("timer canceled message");
+      } else {
+        cb(...args);
+        if (interval) {
+          // TODO Faking setInterval with setTimeout.
+          // We need a new timer implementation, this is just a stopgap.
+          startTimer(id, cb, delay, true, args);
+        }
+      }
+    },
+    error => {
+      throw error;
+    }
+  );
 }
 
 export function setTimeout(
@@ -71,7 +72,9 @@ export function setTimeout(
   // tslint:disable-next-line:no-any
   ...args: any[]
 ): number {
-  return startTimer(cb, delay, false, args);
+  const id = nextTimerId++;
+  startTimer(id, cb, delay, false, args);
+  return id;
 }
 
 export function setInterval(
@@ -80,12 +83,12 @@ export function setInterval(
   // tslint:disable-next-line:no-any
   ...args: any[]
 ): number {
-  return startTimer(cb, delay, true, args);
+  const id = nextTimerId++;
+  startTimer(id, cb, delay, true, args);
+  return id;
 }
 
 export function clearTimer(id: number) {
-  timers.delete(id);
-
   const builder = new flatbuffers.Builder();
   fbs.TimerClear.startTimerClear(builder);
   fbs.TimerClear.addId(builder, id);
diff --git a/js/util.ts b/js/util.ts
index 6971ca1b9..efe0dcf84 100644
--- a/js/util.ts
+++ b/js/util.ts
@@ -83,3 +83,11 @@ export function notImplemented(): never {
 export function unreachable(): never {
   throw new Error("Code not reachable");
 }
+
+export function hexdump(u8: Uint8Array): string {
+  return Array.prototype.map
+    .call(u8, (x: number) => {
+      return ("00" + x.toString(16)).slice(-2);
+    })
+    .join(" ");
+}