deno通信实现的方法（附代码）

interface Libdeno {
 recv(cb: MessageCallback): void;

 send(control: ArrayBufferView, data?: ArrayBufferView): null | Uint8Array;

 print(x: string, isErr?: boolean): void;

 shared: ArrayBuffer;

 /** Evaluate provided code in the current context.
 * It differs from eval(...) in that it does not create a new context.
 * Returns an array: [output, errInfo].
 * If an error occurs, `output` becomes null and `errInfo` is non-null.
 */
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 evalContext(code: string): [any, EvalErrorInfo | null];

 errorToJSON: (e: Error) => string;
}

export const libdeno = window.libdeno as Libdeno;

import { libdeno } from "./libdeno";
function sendInternal(
 builder: flatbuffers.Builder,
 innerType: msg.Any,
 inner: flatbuffers.Offset,
 data: undefined | ArrayBufferView,
 sync = true
): [number, null | Uint8Array] {
 const cmdId = nextCmdId++;
 msg.Base.startBase(builder);
 msg.Base.addInner(builder, inner);
 msg.Base.addInnerType(builder, innerType);
 msg.Base.addSync(builder, sync);
 msg.Base.addCmdId(builder, cmdId);
 builder.finish(msg.Base.endBase(builder));
 const res = libdeno.send(builder.asUint8Array(), data);
 builder.inUse = false;
 return [cmdId, res];
}

const [cmdId, resBuf] = sendInternal(builder, innerType, inner, data, true);

const [cmdId, resBuf] = sendInternal(builder, innerType, inner, data, false);
 util.assert(resBuf == null);
 const promise = util.createResolvable<msg.Base>();
 promiseTable.set(cmdId, promise);
 return promise;

pub fn dispatch(
 isolate: &Isolate,
 control: libdeno::deno_buf,
 data: libdeno::deno_buf,
) -> (bool, Box<Op>) {
 let base = msg::get_root_as_base(&control);
 let is_sync = base.sync();
 let inner_type = base.inner_type();
 let cmd_id = base.cmd_id();

 let op: Box<Op> = if inner_type == msg::Any::SetTimeout {
 // SetTimeout is an exceptional op: the global timeout field is part of the
 // Isolate state (not the IsolateState state) and it must be updated on the
 // main thread.
 assert_eq!(is_sync, true);
 op_set_timeout(isolate, &base, data)
 } else {
 // Handle regular ops.
 let op_creator: OpCreator = match inner_type {
 msg::Any::Accept => op_accept,
 msg::Any::Chdir => op_chdir,
 msg::Any::Chmod => op_chmod,
 msg::Any::Close => op_close,
 msg::Any::FetchModuleMetaData => op_fetch_module_meta_data,
 msg::Any::CopyFile => op_copy_file,
 msg::Any::Cwd => op_cwd,
 msg::Any::Dial => op_dial,
 msg::Any::Environ => op_env,
 msg::Any::Exit => op_exit,
 msg::Any::Fetch => op_fetch,
 msg::Any::FormatError => op_format_error,
 msg::Any::Listen => op_listen,
 msg::Any::MakeTempDir => op_make_temp_dir,
 msg::Any::Metrics => op_metrics,
 msg::Any::Mkdir => op_mkdir,
 msg::Any::Open => op_open,
 msg::Any::ReadDir => op_read_dir,
 msg::Any::ReadFile => op_read_file,
 msg::Any::Readlink => op_read_link,
 msg::Any::Read => op_read,
 msg::Any::Remove => op_remove,
 msg::Any::Rename => op_rename,
 msg::Any::ReplReadline => op_repl_readline,
 msg::Any::ReplStart => op_repl_start,
 msg::Any::Resources => op_resources,
 msg::Any::Run => op_run,
 msg::Any::RunStatus => op_run_status,
 msg::Any::SetEnv => op_set_env,
 msg::Any::Shutdown => op_shutdown,
 msg::Any::Start => op_start,
 msg::Any::Stat => op_stat,
 msg::Any::Symlink => op_symlink,
 msg::Any::Truncate => op_truncate,
 msg::Any::WorkerGetMessage => op_worker_get_message,
 msg::Any::WorkerPostMessage => op_worker_post_message,
 msg::Any::Write => op_write,
 msg::Any::WriteFile => op_write_file,
 msg::Any::Now => op_now,
 msg::Any::IsTTY => op_is_tty,
 msg::Any::Seek => op_seek,
 msg::Any::Permissions => op_permissions,
 msg::Any::PermissionRevoke => op_revoke_permission,
 _ => panic!(format!(
 "Unhandled message {}",
 msg::enum_name_any(inner_type)
 )),
 };
 op_creator(&isolate, &base, data)
 };

 // ...省略多余的代码
}

let (is_sync, op) = dispatch(isolate, control_buf, zero_copy_buf);

 let task = op
 .and_then(move |buf| {
 let sender = tx; // tx is moved to new thread
 sender.send((zero_copy_id, buf)).expect("tx.send error");
 Ok(())
 }).map_err(|_| ());
 tokio::spawn(task);

let (tx, rx) = mpsc::channel::<(usize, Buf)>();

pub fn event_loop(&self) -> Result<(), JSError> {
 // Main thread event loop.
 while !self.is_idle() {
 match recv_deadline(&self.rx, self.get_timeout_due()) {
 Ok((zero_copy_id, buf)) => self.complete_op(zero_copy_id, buf),
 Err(mpsc::RecvTimeoutError::Timeout) => self.timeout(),
 Err(e) => panic!("recv_deadline() failed: {:?}", e),
 }
 self.check_promise_errors();
 if let Some(err) = self.last_exception() {
 return Err(err);
 }
 }
 // Check on done
 self.check_promise_errors();
 if let Some(err) = self.last_exception() {
 return Err(err);
 }
 Ok(())
 }

fn is_idle(&self) -> bool {
 self.ntasks.get() == 0 && self.get_timeout_due().is_none()
 }

fn ntasks_increment(&self) {
 assert!(self.ntasks.get() >= 0);
 self.ntasks.set(self.ntasks.get() + 1);
 }

export function start(source?: string): msg.StartRes {
 libdeno.recv(handleAsyncMsgFromRust);

 // First we send an empty `Start` message to let the privileged side know we
 // are ready. The response should be a `StartRes` message containing the CLI
 // args and other info.
 const startResMsg = sendStart();

 util.setLogDebug(startResMsg.debugFlag(), source);

 setGlobals(startResMsg.pid(), startResMsg.noColor(), startResMsg.execPath()!);

 return startResMsg;
}

export function handleAsyncMsgFromRust(ui8: Uint8Array): void {
 // If a the buffer is empty, recv() on the native side timed out and we
 // did not receive a message.
 if (ui8 && ui8.length) {
 const bb = new flatbuffers.ByteBuffer(ui8);
 const base = msg.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);
 }
 }
 // Fire timers that have become runnable.
 fireTimers();
}