Map promises onto futures.

Refactors handlers.rs

The idea is that all Deno "ops" (aka bindings) should map onto
a Rust Future. By setting the "sync" flag in the Base message
users can determine if the future is executed immediately or put
on the event loop.

In the case of async futures, a promise is automatically created.
Errors are automatically forwarded and raised.

TODO:

- The file system ops in src/handler.rs are not using the thread pool
  yet. This will be done in the future using tokio_threadpool::blocking.
  That is, if you try to call them asynchronously, you will get a promise
  and it will act asynchronous, but currently it will be blocking.
- Handlers in src/handler.rs returned boxed futures. This was to make
  it easy while developing. We should try to remove this allocation.

1 files changed, 59 insertions, 9 deletions

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())];
+}