feat: MessageChannel and MessagePort

This commit introduces support for MessageChannel and MessagePort.
MessagePorts can be transfered across a message channel.

cli/dts/lib.deno.shared_globals.d.ts

@@ -382,10 +382,6 @@ declare class ErrorEvent extends Event {
   constructor(type: string, eventInitDict?: ErrorEventInit);
 }
 
-interface PostMessageOptions {
-  transfer?: any[];
-}
-
 interface AbstractWorkerEventMap {
   "error": ErrorEvent;
 }

extensions/web/02_event.js

@@ -1129,6 +1129,7 @@
       });
 
       this.data = eventInitDict?.data ?? null;
+      this.ports = eventInitDict?.ports ?? [];
       this.origin = eventInitDict?.origin ?? "";
       this.lastEventId = eventInitDict?.lastEventId ?? "";
     }

extensions/web/13_message_port.js

@@ -0,0 +1,180 @@
+// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
+
+// @ts-check
+/// <reference path="../../core/lib.deno_core.d.ts" />
+/// <reference path="../webidl/internal.d.ts" />
+/// <reference path="./internal.d.ts" />
+/// <reference path="./lib.deno_web.d.ts" />
+
+"use strict";
+
+((window) => {
+  const core = window.Deno.core;
+  const webidl = window.__bootstrap.webidl;
+  const { setEventTargetData } = window.__bootstrap.eventTarget;
+
+  class MessageChannel {
+    /** @type {MessagePort} */
+    #port1;
+    /** @type {MessagePort} */
+    #port2;
+
+    constructor() {
+      this[webidl.brand] = webidl.brand;
+      const [port1Id, port2Id] = opCreateEntangledMessagePort();
+      const port1 = createMessagePort(port1Id);
+      const port2 = createMessagePort(port2Id);
+      this.#port1 = port1;
+      this.#port2 = port2;
+    }
+
+    get port1() {
+      webidl.assertBranded(this, MessageChannel);
+      return this.#port1;
+    }
+
+    get port2() {
+      webidl.assertBranded(this, MessageChannel);
+      return this.#port2;
+    }
+
+    [Symbol.for("Deno.inspect")](inspect) {
+      return `MessageChannel ${
+        inspect({ port1: this.port1, port2: this.port2 })
+      }`;
+    }
+
+    get [Symbol.toStringTag]() {
+      return "MessageChannel";
+    }
+  }
+
+  const _id = Symbol("id");
+
+  /**
+   * @param {number} id
+   * @returns {MessagePort}
+   */
+  function createMessagePort(id) {
+    const port = webidl.createBranded(MessagePort);
+    setEventTargetData(port);
+    port[_id] = id;
+    return port;
+  }
+
+  class MessagePort extends EventTarget {
+    /** @type {number | null} */
+    [_id];
+
+    constructor() {
+      super();
+      webidl.illegalConstructor();
+    }
+
+    /**
+     * @param {any} message
+     * @param {object[] | PostMessageOptions} transferOrOptions
+     */
+    postMessage(message, transferOrOptions = {}) {
+      let transfer = [];
+      if (Array.isArray(transferOrOptions)) {
+        transfer = transferOrOptions;
+      } else if (Array.isArray(transferOrOptions.transfer)) {
+        transfer = transferOrOptions.transfer;
+      }
+
+      const data = serializeJsMessageData(message, transfer);
+      core.opSync("op_message_port_post_message", this[_id], data);
+    }
+
+    start() {
+      const self = this;
+      (async () => {
+        while (true) {
+          const data = await core.opAsync(
+            "op_message_port_recv_message",
+            this[_id],
+          );
+          if (data === null) break;
+          const [message, transfer] = deserializeJsMessageData(data);
+          const event = new MessageEvent("message", { data: message, ports: transfer });
+          self.dispatchEvent(event);
+        }
+      })();
+    }
+
+    close() {
+      if (this[_id] !== null) {
+        core.close(this[_id]);
+      }
+    }
+  }
+
+  /**
+   * @returns {[number, number]}
+   */
+  function opCreateEntangledMessagePort() {
+    return core.opSync("op_message_port_create_entangled");
+  }
+
+  /**
+   * @param {globalThis.__bootstrap.messagePort.MessageData} messageData
+   * @returns {[any, object[]]}
+   */
+  function deserializeJsMessageData(messageData) {
+    /** @type {object[]} */
+    const transferables = [];
+
+    for (const transferable of messageData.transferables) {
+      switch (transferable.kind) {
+        case "messagePort":
+          const port = createMessagePort(transferable.data);
+          transferables.push(port);
+          break;
+        default:
+          throw new TypeError("Unreachable");
+      }
+    }
+
+    const data = core.deserialize(messageData.data);
+
+    return [data, transferables];
+  }
+
+  /**
+   * @param {any} data
+   * @param {object[]} tranferables
+   * @returns {globalThis.__bootstrap.messagePort.MessageData}
+   */
+  function serializeJsMessageData(data, tranferables) {
+    const serializedData = core.serialize(data);
+
+    /** @type {globalThis.__bootstrap.messagePort.Transferable[]} */
+    const serializedTransferables = [];
+
+    for (const transferable of tranferables) {
+      if (transferable instanceof MessagePort) {
+        webidl.assertBranded(transferable, MessagePort);
+        const id = transferable[_id];
+        if (id === null) {
+          throw new TypeError("Can not transfer disentangled message port");
+        }
+        serializedTransferables.push({ kind: "messagePort", data: id });
+      } else {
+        throw new TypeError("Value not transferable");
+      }
+    }
+
+    return {
+      data: serializedData,
+      transferables: serializedTransferables,
+    };
+  }
+
+  window.__bootstrap.messagePort = {
+    MessageChannel,
+    MessagePort,
+    deserializeJsMessageData,
+    serializeJsMessageData,
+  };
+})(globalThis);

extensions/web/Cargo.toml

@@ -18,6 +18,7 @@ base64 = "0.13.0"
 deno_core = { version = "0.90.0", path = "../../core" }
 encoding_rs = "0.8.28"
 serde = "1.0"
+tokio = "1.7"
 uuid = { version = "0.8.2", features = ["v4"] }
 
 [dev-dependencies]

extensions/web/internal.d.ts

@@ -4,9 +4,6 @@
 /// <reference lib="esnext" />
 
 declare namespace globalThis {
-  declare var TextEncoder: typeof TextEncoder;
-  declare var TextDecoder: typeof TextDecoder;
-
   declare namespace __bootstrap {
     declare var infra: {
       collectSequenceOfCodepoints(
@@ -85,5 +82,16 @@ declare namespace globalThis {
       ReadableStream: typeof ReadableStream;
       isReadableStreamDisturbed(stream: ReadableStream): boolean;
     };
+
+    declare namespace messagePort {
+      declare type Transferable = {
+        kind: "messagePort";
+        data: number;
+      };
+      declare interface MessageData {
+        data: Uint8Array;
+        transferables: Transferable[];
+      }
+    }
   }
 }

extensions/web/lib.deno_web.d.ts

@@ -648,3 +648,7 @@ interface TransformStreamDefaultControllerTransformCallback<I, O> {
     controller: TransformStreamDefaultController<O>,
   ): void | PromiseLike<void>;
 }
+
+interface PostMessageOptions {
+  transfer?: any[];
+}

extensions/web/lib.rs

@@ -1,11 +1,16 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
 
+mod message_port;
+
+pub use crate::message_port::JsMessageData;
+
 use deno_core::error::bad_resource_id;
 use deno_core::error::null_opbuf;
 use deno_core::error::range_error;
 use deno_core::error::type_error;
 use deno_core::error::AnyError;
 use deno_core::include_js_files;
+use deno_core::op_async;
 use deno_core::op_sync;
 use deno_core::url::Url;
 use deno_core::Extension;
@@ -30,6 +35,10 @@ use std::sync::Mutex;
 use std::usize;
 use uuid::Uuid;
 
+use crate::message_port::op_message_port_create_entangled;
+use crate::message_port::op_message_port_post_message;
+use crate::message_port::op_message_port_recv_message;
+
 /// Load and execute the javascript code.
 pub fn init(
   blob_url_store: BlobUrlStore,
@@ -52,6 +61,7 @@ pub fn init(
       "10_filereader.js",
       "11_blob_url.js",
       "12_location.js",
+      "13_message_port.js",
     ))
     .ops(vec![
       ("op_base64_decode", op_sync(op_base64_decode)),
@@ -71,6 +81,18 @@ pub fn init(
         "op_file_revoke_object_url",
         op_sync(op_file_revoke_object_url),
       ),
+      (
+        "op_message_port_create_entangled",
+        op_sync(op_message_port_create_entangled),
+      ),
+      (
+        "op_message_port_post_message",
+        op_sync(op_message_port_post_message),
+      ),
+      (
+        "op_message_port_recv_message",
+        op_async(op_message_port_recv_message),
+      ),
     ])
     .state(move |state| {
       state.put(blob_url_store.clone());