feat(ext/web): EventSource (#14730)

Closes #10298

---------

Co-authored-by: Aapo Alasuutari <aapo.alasuutari@gmail.com>

cli/tests/node_compat/test/common/index.js

@@ -37,6 +37,7 @@ let knownGlobals = [
   crypto,
   Deno,
   dispatchEvent,
+  EventSource,
   fetch,
   getParent,
   global,

ext/fetch/26_fetch.js

@@ -594,4 +594,4 @@ function handleWasmStreaming(source, rid) {
   }
 }
 
-export { fetch, handleWasmStreaming };
+export { fetch, handleWasmStreaming, mainFetch };

ext/fetch/27_eventsource.js

@@ -0,0 +1,379 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+/// <reference path="../../core/internal.d.ts" />
+
+const core = globalThis.Deno.core;
+
+import * as webidl from "ext:deno_webidl/00_webidl.js";
+import { URL } from "ext:deno_url/00_url.js";
+import DOMException from "ext:deno_web/01_dom_exception.js";
+import {
+  defineEventHandler,
+  EventTarget,
+  setIsTrusted,
+} from "ext:deno_web/02_event.js";
+import { TransformStream } from "ext:deno_web/06_streams.js";
+import { TextDecoderStream } from "ext:deno_web/08_text_encoding.js";
+import { getLocationHref } from "ext:deno_web/12_location.js";
+import { newInnerRequest } from "ext:deno_fetch/23_request.js";
+import { mainFetch } from "ext:deno_fetch/26_fetch.js";
+
+const primordials = globalThis.__bootstrap.primordials;
+const {
+  ArrayPrototypeFind,
+  Number,
+  NumberIsFinite,
+  NumberIsNaN,
+  ObjectDefineProperties,
+  Promise,
+  StringPrototypeEndsWith,
+  StringPrototypeIncludes,
+  StringPrototypeIndexOf,
+  StringPrototypeSlice,
+  StringPrototypeSplit,
+  StringPrototypeStartsWith,
+  StringPrototypeToLowerCase,
+  Symbol,
+} = primordials;
+
+// Copied from https://github.com/denoland/deno_std/blob/e0753abe0c8602552862a568348c046996709521/streams/text_line_stream.ts#L20-L74
+export class TextLineStream extends TransformStream {
+  #allowCR;
+  #buf = "";
+
+  constructor(options) {
+    super({
+      transform: (chunk, controller) => this.#handle(chunk, controller),
+      flush: (controller) => {
+        if (this.#buf.length > 0) {
+          if (
+            this.#allowCR &&
+            this.#buf[this.#buf.length - 1] === "\r"
+          ) controller.enqueue(StringPrototypeSlice(this.#buf, 0, -1));
+          else controller.enqueue(this.#buf);
+        }
+      },
+    });
+    this.#allowCR = options?.allowCR ?? false;
+  }
+
+  #handle(chunk, controller) {
+    chunk = this.#buf + chunk;
+
+    for (;;) {
+      const lfIndex = StringPrototypeIndexOf(chunk, "\n");
+
+      if (this.#allowCR) {
+        const crIndex = StringPrototypeIndexOf(chunk, "\r");
+
+        if (
+          crIndex !== -1 && crIndex !== (chunk.length - 1) &&
+          (lfIndex === -1 || (lfIndex - 1) > crIndex)
+        ) {
+          controller.enqueue(StringPrototypeSlice(chunk, 0, crIndex));
+          chunk = StringPrototypeSlice(chunk, crIndex + 1);
+          continue;
+        }
+      }
+
+      if (lfIndex !== -1) {
+        let crOrLfIndex = lfIndex;
+        if (chunk[lfIndex - 1] === "\r") {
+          crOrLfIndex--;
+        }
+        controller.enqueue(StringPrototypeSlice(chunk, 0, crOrLfIndex));
+        chunk = StringPrototypeSlice(chunk, lfIndex + 1);
+        continue;
+      }
+
+      break;
+    }
+
+    this.#buf = chunk;
+  }
+}
+
+const CONNECTING = 0;
+const OPEN = 1;
+const CLOSED = 2;
+
+const _url = Symbol("[[url]]");
+const _withCredentials = Symbol("[[withCredentials]]");
+const _readyState = Symbol("[[readyState]]");
+const _reconnectionTime = Symbol("[[reconnectionTime]]");
+const _lastEventID = Symbol("[[lastEventID]]");
+const _abortController = Symbol("[[abortController]]");
+const _loop = Symbol("[[loop]]");
+
+class EventSource extends EventTarget {
+  /** @type {AbortController} */
+  [_abortController] = new AbortController();
+
+  /** @type {number} */
+  [_reconnectionTime] = 5000;
+
+  /** @type {string} */
+  [_lastEventID] = "";
+
+  /** @type {number} */
+  [_readyState] = CONNECTING;
+  get readyState() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    return this[_readyState];
+  }
+
+  get CONNECTING() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    return CONNECTING;
+  }
+  get OPEN() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    return OPEN;
+  }
+  get CLOSED() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    return CLOSED;
+  }
+
+  /** @type {string} */
+  [_url];
+  get url() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    return this[_url];
+  }
+
+  /** @type {boolean} */
+  [_withCredentials];
+  get withCredentials() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    return this[_withCredentials];
+  }
+
+  constructor(url, eventSourceInitDict = {}) {
+    super();
+    this[webidl.brand] = webidl.brand;
+    const prefix = "Failed to construct 'EventSource'";
+    webidl.requiredArguments(arguments.length, 1, {
+      prefix,
+    });
+    url = webidl.converters.USVString(url, {
+      prefix,
+      context: "Argument 1",
+    });
+    eventSourceInitDict = webidl.converters.EventSourceInit(
+      eventSourceInitDict,
+      {
+        prefix,
+        context: "Argument 2",
+      },
+    );
+
+    try {
+      url = new URL(url, getLocationHref()).href;
+    } catch (e) {
+      throw new DOMException(e.message, "SyntaxError");
+    }
+
+    this[_url] = url;
+    this[_withCredentials] = eventSourceInitDict.withCredentials;
+
+    this[_loop]();
+  }
+
+  close() {
+    webidl.assertBranded(this, EventSourcePrototype);
+    this[_abortController].abort();
+    this[_readyState] = CLOSED;
+  }
+
+  async [_loop]() {
+    let lastEventIDValue = "";
+    while (this[_readyState] !== CLOSED) {
+      const lastEventIDValueCopy = lastEventIDValue;
+      lastEventIDValue = "";
+      const req = newInnerRequest(
+        "GET",
+        this[_url],
+        () =>
+          lastEventIDValueCopy === ""
+            ? [
+              ["accept", "text/event-stream"],
+            ]
+            : [
+              ["accept", "text/event-stream"],
+              [
+                "Last-Event-Id",
+                core.ops.op_utf8_to_byte_string(lastEventIDValueCopy),
+              ],
+            ],
+        null,
+        false,
+      );
+      /** @type {InnerResponse} */
+      const res = await mainFetch(req, true, this[_abortController].signal);
+
+      const contentType = ArrayPrototypeFind(
+        res.headerList,
+        (header) => StringPrototypeToLowerCase(header[0]) === "content-type",
+      );
+      if (res.type === "error") {
+        if (res.aborted) {
+          this[_readyState] = CLOSED;
+          this.dispatchEvent(new Event("error"));
+          break;
+        } else {
+          if (this[_readyState] === CLOSED) {
+            this[_abortController].abort();
+            break;
+          }
+          this[_readyState] = CONNECTING;
+          this.dispatchEvent(new Event("error"));
+          await new Promise((res) => setTimeout(res, this[_reconnectionTime]));
+          if (this[_readyState] !== CONNECTING) {
+            continue;
+          }
+
+          if (this[_lastEventID] !== "") {
+            lastEventIDValue = this[_lastEventID];
+          }
+          continue;
+        }
+      } else if (
+        res.status !== 200 ||
+        !StringPrototypeIncludes(
+          contentType?.[1].toLowerCase(),
+          "text/event-stream",
+        )
+      ) {
+        this[_readyState] = CLOSED;
+        this.dispatchEvent(new Event("error"));
+        break;
+      }
+
+      if (this[_readyState] !== CLOSED) {
+        this[_readyState] = OPEN;
+        this.dispatchEvent(new Event("open"));
+
+        let data = "";
+        let eventType = "";
+        let lastEventID = this[_lastEventID];
+
+        for await (
+          // deno-lint-ignore prefer-primordials
+          const chunk of res.body.stream
+            .pipeThrough(new TextDecoderStream())
+            .pipeThrough(new TextLineStream({ allowCR: true }))
+        ) {
+          if (chunk === "") {
+            this[_lastEventID] = lastEventID;
+            if (data === "") {
+              eventType = "";
+              continue;
+            }
+            if (StringPrototypeEndsWith(data, "\n")) {
+              data = StringPrototypeSlice(data, 0, -1);
+            }
+            const event = new MessageEvent(eventType || "message", {
+              data,
+              origin: res.url(),
+              lastEventId: this[_lastEventID],
+            });
+            setIsTrusted(event, true);
+            data = "";
+            eventType = "";
+            if (this[_readyState] !== CLOSED) {
+              this.dispatchEvent(event);
+            }
+          } else if (StringPrototypeStartsWith(chunk, ":")) {
+            continue;
+          } else {
+            let field = chunk;
+            let value = "";
+            if (StringPrototypeIncludes(chunk, ":")) {
+              ({ 0: field, 1: value } = StringPrototypeSplit(chunk, ":"));
+              if (StringPrototypeStartsWith(value, " ")) {
+                value = StringPrototypeSlice(value, 1);
+              }
+            }
+
+            switch (field) {
+              case "event": {
+                eventType = value;
+                break;
+              }
+              case "data": {
+                data += value + "\n";
+                break;
+              }
+              case "id": {
+                if (!StringPrototypeIncludes(value, "\0")) {
+                  lastEventID = value;
+                }
+                break;
+              }
+              case "retry": {
+                const reconnectionTime = Number(value);
+                if (
+                  !NumberIsNaN(reconnectionTime) &&
+                  NumberIsFinite(reconnectionTime)
+                ) {
+                  this[_reconnectionTime] = reconnectionTime;
+                }
+                break;
+              }
+            }
+          }
+
+          if (this[_abortController].signal.aborted) {
+            break;
+          }
+        }
+        if (this[_readyState] === CLOSED) {
+          this[_abortController].abort();
+          break;
+        }
+        this[_readyState] = CONNECTING;
+        this.dispatchEvent(new Event("error"));
+        await new Promise((res) => setTimeout(res, this[_reconnectionTime]));
+        if (this[_readyState] !== CONNECTING) {
+          continue;
+        }
+
+        if (this[_lastEventID] !== "") {
+          lastEventIDValue = this[_lastEventID];
+        }
+      }
+    }
+  }
+}
+
+const EventSourcePrototype = EventSource.prototype;
+
+ObjectDefineProperties(EventSource, {
+  CONNECTING: {
+    value: 0,
+  },
+  OPEN: {
+    value: 1,
+  },
+  CLOSED: {
+    value: 2,
+  },
+});
+
+defineEventHandler(EventSource.prototype, "open");
+defineEventHandler(EventSource.prototype, "message");
+defineEventHandler(EventSource.prototype, "error");
+
+webidl.converters.EventSourceInit = webidl.createDictionaryConverter(
+  "EventSourceInit",
+  [
+    {
+      key: "withCredentials",
+      defaultValue: false,
+      converter: webidl.converters.boolean,
+    },
+  ],
+);
+
+export { EventSource };

ext/fetch/lib.rs

@@ -112,6 +112,7 @@ deno_core::extension!(deno_fetch,
     op_fetch<FP>,
     op_fetch_send,
     op_fetch_response_upgrade,
+    op_utf8_to_byte_string,
     op_fetch_custom_client<FP>,
   ],
   esm = [
@@ -121,7 +122,8 @@ deno_core::extension!(deno_fetch,
     "22_http_client.js",
     "23_request.js",
     "23_response.js",
-    "26_fetch.js"
+    "26_fetch.js",
+    "27_eventsource.js"
   ],
   options = {
     options: Options,
@@ -969,3 +971,11 @@ pub fn create_http_client(
 
   builder.build().map_err(|e| e.into())
 }
+
+#[op2]
+#[serde]
+pub fn op_utf8_to_byte_string(
+  #[string] input: String,
+) -> Result<ByteString, AnyError> {
+  Ok(input.into())
+}

runtime/js/98_global_scope.js

@@ -32,6 +32,7 @@ import * as formData from "ext:deno_fetch/21_formdata.js";
 import * as request from "ext:deno_fetch/23_request.js";
 import * as response from "ext:deno_fetch/23_response.js";
 import * as fetch from "ext:deno_fetch/26_fetch.js";
+import * as eventSource from "ext:deno_fetch/27_eventsource.js";
 import * as messagePort from "ext:deno_web/13_message_port.js";
 import * as webidl from "ext:deno_webidl/00_webidl.js";
 import DOMException from "ext:deno_web/01_dom_exception.js";
@@ -129,6 +130,7 @@ const windowOrWorkerGlobalScope = {
   Crypto: util.nonEnumerable(crypto.Crypto),
   SubtleCrypto: util.nonEnumerable(crypto.SubtleCrypto),
   fetch: util.writable(fetch.fetch),
+  EventSource: util.writable(eventSource.EventSource),
   performance: util.writable(performance.performance),
   reportError: util.writable(event.reportError),
   setInterval: util.writable(timers.setInterval),

tools/wpt/expectation.json

@@ -8211,7 +8211,6 @@
         ],
         "expected-self-properties.worker.html": [
           "existence of XMLHttpRequest",
-          "existence of EventSource",
           "existence of SharedWorker"
         ],
         "unexpected-self-properties.worker.html": true
@@ -8302,7 +8301,6 @@
           "The CanvasPath interface object should be exposed.",
           "The TextMetrics interface object should be exposed.",
           "The Path2D interface object should be exposed.",
-          "The EventSource interface object should be exposed.",
           "The XMLHttpRequestEventTarget interface object should be exposed.",
           "The XMLHttpRequestUpload interface object should be exposed.",
           "The XMLHttpRequest interface object should be exposed.",
@@ -10643,5 +10641,91 @@
     "media": {
       "media-sniff.window.html": false
     }
+  },
+  "eventsource": {
+    "dedicated-worker": {
+      "eventsource-eventtarget.worker.html": true,
+      "eventsource-constructor-no-new.any.html": true,
+      "eventsource-constructor-no-new.any.worker.html": true
+    },
+    "event-data.any.html": true,
+    "event-data.any.worker.html": true,
+    "eventsource-constructor-empty-url.any.html": true,
+    "eventsource-constructor-empty-url.any.worker.html": true,
+    "eventsource-constructor-url-bogus.any.html": true,
+    "eventsource-constructor-url-bogus.any.worker.html": true,
+    "eventsource-eventtarget.any.html": true,
+    "eventsource-eventtarget.any.worker.html": true,
+    "eventsource-onmessage-trusted.any.html": true,
+    "eventsource-onmessage-trusted.any.worker.html": true,
+    "eventsource-onmessage.any.html": true,
+    "eventsource-onmessage.any.worker.html": true,
+    "eventsource-onopen.any.html": true,
+    "eventsource-onopen.any.worker.html": true,
+    "eventsource-prototype.any.html": true,
+    "eventsource-prototype.any.worker.html": true,
+    "eventsource-request-cancellation.window.any.html": false,
+    "eventsource-request-cancellation.window.any.worker.html": false,
+    "eventsource-url.any.html": true,
+    "eventsource-url.any.worker.html": true,
+    "format-bom-2.any.html": true,
+    "format-bom-2.any.worker.html": true,
+    "format-bom.any.html": true,
+    "format-bom.any.worker.html": true,
+    "format-comments.any.html": true,
+    "format-comments.any.worker.html": true,
+    "format-data-before-final-empty-line.any.html": true,
+    "format-data-before-final-empty-line.any.worker.html": true,
+    "format-field-data.any.html": true,
+    "format-field-data.any.worker.html": true,
+    "format-field-event-empty.any.html": true,
+    "format-field-event-empty.any.worker.html": true,
+    "format-field-event.any.html": true,
+    "format-field-event.any.worker.html": true,
+    "format-field-id-2.any.html": true,
+    "format-field-id-2.any.worker.html": true,
+    "format-field-id-3.window.html": true,
+    "format-field-id-null.window.html": true,
+    "format-field-id.any.html": true,
+    "format-field-id.any.worker.html": true,
+    "format-field-parsing.any.html": true,
+    "format-field-parsing.any.worker.html": true,
+    "format-field-retry-bogus.any.html": true,
+    "format-field-retry-bogus.any.worker.html": true,
+    "format-field-retry-empty.any.html": true,
+    "format-field-retry-empty.any.worker.html": true,
+    "format-field-retry.any.html": true,
+    "format-field-retry.any.worker.html": true,
+    "format-field-unknown.any.html": true,
+    "format-field-unknown.any.worker.html": true,
+    "format-leading-space.any.html": true,
+    "format-leading-space.any.worker.html": true,
+    "format-mime-bogus.any.html": true,
+    "format-mime-bogus.any.worker.html": true,
+    "format-mime-trailing-semicolon.any.html": true,
+    "format-mime-trailing-semicolon.any.worker.html": true,
+    "format-mime-valid-bogus.any.html": true,
+    "format-mime-valid-bogus.any.worker.html": true,
+    "format-newlines.any.html": true,
+    "format-newlines.any.worker.html": true,
+    "format-null-character.any.html": true,
+    "format-null-character.any.worker.html": true,
+    "format-utf-8.any.html": true,
+    "format-utf-8.any.worker.html": true,
+    "request-accept.any.html": true,
+    "request-accept.any.worker.html": true,
+    "request-cache-control.any.html": false,
+    "request-cache-control.any.worker.html": false,
+    "request-credentials.window.any.html": false,
+    "request-credentials.window.any.worker.html": false,
+    "request-redirect.window.any.html": false,
+    "request-redirect.window.any.worker.html": false,
+    "eventsource-close.window.html": false,
+    "eventsource-constructor-document-domain.window.html": false,
+    "eventsource-constructor-non-same-origin.window.html": false,
+    "eventsource-constructor-stringify.window.html": false,
+    "eventsource-cross-origin.window.html": false,
+    "eventsource-reconnect.window.html": false,
+    "request-status-error.window.html": false
   }
-}
+}

tools/wpt/runner.ts

@@ -186,7 +186,10 @@ async function generateBundle(location: URL): Promise<string> {
   if (title) {
     const url = new URL(`#${inlineScriptCount}`, location);
     inlineScriptCount++;
-    scriptContents.push([url.href, `globalThis.META_TITLE="${title}"`]);
+    scriptContents.push([
+      url.href,
+      `globalThis.META_TITLE=${JSON.stringify(title)}`,
+    ]);
   }
   for (const script of scripts) {
     const src = script.getAttribute("src");