ext/fetch/27_eventsource.js - main - githubmirror/deno

ext/fetch/27_eventsource.js
452 строки12 KB
Kenta Moriuchi
chore: make prefer-primordials lint an internal plugin (#35978)
23 июл 2026, 11:53
23 июл 2026, 11:535b55d08
// Copyright 2018-2026 the Deno authors. MIT license.

(function () {
const { core, primordials } = __bootstrap;
const { op_utf8_to_byte_string } = core.ops;
const {
  ArrayPrototypeFind,
  ArrayPrototypeJoin,
  ArrayPrototypePush,
  Number,
  NumberIsFinite,
  NumberIsNaN,
  ObjectDefineProperties,
  ObjectPrototypeIsPrototypeOf,
  StringPrototypeEndsWith,
  StringPrototypeIncludes,
  StringPrototypeIndexOf,
  StringPrototypeSlice,
  StringPrototypeStartsWith,
  StringPrototypeToLowerCase,
  SymbolFor,
} = primordials;

const webidl = core.loadExtScript("ext:deno_webidl/00_webidl.js");
const { createFilteredInspectProxy } = core.loadExtScript(
  "ext:deno_web/01_console.js",
);
const { URL } = core.loadExtScript("ext:deno_web/00_url.js");
const { DOMException } = core.loadExtScript("ext:deno_web/01_dom_exception.js");
const {
  defineEventHandler,
  EventTarget,
  setIsTrusted,
} = core.loadExtScript("ext:deno_web/02_event.js");
const { TransformStream } = core.loadExtScript("ext:deno_web/06_streams.js");
const { TextDecoderStream } = core.loadExtScript(
  "ext:deno_web/08_text_encoding.js",
);
const { getLocationHref } = core.loadExtScript("ext:deno_web/12_location.js");
const { newInnerRequest } = core.loadExtScript("ext:deno_fetch/23_request.js");
const { mainFetch } = core.loadExtScript("ext:deno_fetch/26_fetch.js");
const {
  fillHeaders,
  headerListFromHeaders,
  headersFromHeaderList,
} = core.loadExtScript("ext:deno_fetch/20_headers.js");

// Same semantics as
// https://github.com/denoland/deno_std/blob/e0753abe0c8602552862a568348c046996709521/streams/text_line_stream.ts#L20-L74
// but linear in the input size: fragments that cannot complete a line are
// buffered in an array and joined only when a line terminator arrives, and
// complete lines are extracted by index scanning instead of re-slicing, so
// a line spanning many fragments costs O(length) rather than O(length^2).
class TextLineStream extends TransformStream {
  #allowCR;
  #frags = [];

  constructor(options) {
    super({
      transform: (chunk, controller) => this.#handle(chunk, controller),
      flush: (controller) => {
        if (this.#frags.length > 0) {
          const buf = ArrayPrototypeJoin(this.#frags, "");
          if (this.#allowCR && StringPrototypeEndsWith(buf, "\r")) {
            controller.enqueue(StringPrototypeSlice(buf, 0, -1));
          } else {
            controller.enqueue(buf);
          }
        }
      },
    });
    this.#allowCR = options?.allowCR ?? false;
  }

  #handle(chunk, controller) {
    if (chunk.length === 0) {
      return;
    }
    const frags = this.#frags;
    // Fast path: the fragment cannot complete a line (no LF; with allowCR
    // also no CR, and the buffered tail does not end with a CR that this
    // fragment would turn into a line break). Just buffer it.
    if (
      !StringPrototypeIncludes(chunk, "\n") &&
      (!this.#allowCR ||
        (!StringPrototypeIncludes(chunk, "\r") &&
          (frags.length === 0 ||
            !StringPrototypeEndsWith(frags[frags.length - 1], "\r"))))
    ) {
      ArrayPrototypePush(frags, chunk);
      return;
    }

    let s;
    if (frags.length > 0) {
      ArrayPrototypePush(frags, chunk);
      s = ArrayPrototypeJoin(frags, "");
      frags.length = 0;
    } else {
      s = chunk;
    }

    let start = 0;
    // Cached position of the next CR at or past `start`; refreshed only once
    // `start` moves past it, so CR-less spans are not rescanned per line.
    let crIndex = this.#allowCR ? StringPrototypeIndexOf(s, "\r", start) : -1;
    for (;;) {
      const lfIndex = StringPrototypeIndexOf(s, "\n", start);

      if (this.#allowCR) {
        if (crIndex !== -1 && crIndex < start) {
          crIndex = StringPrototypeIndexOf(s, "\r", start);
        }

        if (
          crIndex !== -1 && crIndex !== (s.length - 1) &&
          (lfIndex === -1 || (lfIndex - 1) > crIndex)
        ) {
          controller.enqueue(StringPrototypeSlice(s, start, crIndex));
          start = crIndex + 1;
          continue;
        }
      }

      if (lfIndex !== -1) {
        let crOrLfIndex = lfIndex;
        if (lfIndex > start && s[lfIndex - 1] === "\r") {
          crOrLfIndex--;
        }
        controller.enqueue(StringPrototypeSlice(s, start, crOrLfIndex));
        start = lfIndex + 1;
        continue;
      }

      break;
    }

    if (start < s.length) {
      ArrayPrototypePush(frags, StringPrototypeSlice(s, start));
    }
  }
}

const CONNECTING = 0;
const OPEN = 1;
const CLOSED = 2;

class EventSource extends EventTarget {
  /** @type {AbortController} */
  #abortController = new AbortController();

  /** @type {number | undefined} */
  #reconnectionTimerId;

  /** @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;
  }

  #headers;

  constructor(url, eventSourceInitDict = { __proto__: null }) {
    super();
    this[webidl.brand] = webidl.brand;
    const prefix = "Failed to construct 'EventSource'";
    webidl.requiredArguments(arguments.length, 1, prefix);
    url = webidl.converters.USVString(url, prefix, "Argument 1");
    eventSourceInitDict = webidl.converters.EventSourceInit(
      eventSourceInitDict,
      prefix,
      "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.#headers = eventSourceInitDict.headers;

    this.#loop();
  }

  close() {
    webidl.assertBranded(this, EventSourcePrototype);
    this.#abortController.abort();
    this.#readyState = CLOSED;
    if (this.#reconnectionTimerId) core.cancelTimer(this.#reconnectionTimerId);
  }

  async #loop() {
    const lastEventIdValue = this.#lastEventId;
    const initialHeaders = headersFromHeaderList(
      lastEventIdValue === ""
        ? [
          ["accept", "text/event-stream"],
        ]
        : [
          ["accept", "text/event-stream"],
          ["Last-Event-Id", op_utf8_to_byte_string(lastEventIdValue)],
        ],
      "request",
    );

    if (this.#headers) {
      fillHeaders(initialHeaders, this.#headers);
    }

    const req = newInnerRequest(
      "GET",
      this.#url,
      () => headerListFromHeaders(initialHeaders),
      null,
      false,
    );

    /** @type {InnerResponse} */
    let res;
    try {
      res = await mainFetch(req, true, this.#abortController.signal);
    } catch {
      this.#reestablishConnection();
      return;
    }

    if (res.aborted) {
      this.#failConnection();
      return;
    }
    if (res.type === "error") {
      this.#reestablishConnection();
      return;
    }
    const contentType = ArrayPrototypeFind(
      res.headerList,
      (header) => StringPrototypeToLowerCase(header[0]) === "content-type",
    );
    if (
      res.status !== 200 ||
      !contentType ||
      !StringPrototypeIncludes(
        StringPrototypeToLowerCase(contentType[1]),
        "text/event-stream",
      )
    ) {
      this.#failConnection();
      return;
    }

    if (this.#readyState === CLOSED) {
      return;
    }
    this.#readyState = OPEN;
    this.dispatchEvent(new Event("open"));

    let data = "";
    let eventType = "";
    let lastEventId = this.#lastEventId;

    try {
      for await (
        // deno-lint-ignore deno-internal/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 = "";
          const colonIndex = StringPrototypeIndexOf(chunk, ":");
          if (colonIndex !== -1) {
            field = StringPrototypeSlice(chunk, 0, colonIndex);
            value = StringPrototypeSlice(chunk, colonIndex + 1);
            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;
            }
          }
        }
      }
    } catch {
      // The connection is reestablished below
    }

    this.#reestablishConnection();
  }

  #reestablishConnection() {
    if (this.#readyState === CLOSED) {
      return;
    }
    this.#readyState = CONNECTING;
    this.dispatchEvent(new Event("error"));
    this.#reconnectionTimerId = core.createSystemTimer(
      () => {
        if (this.#readyState !== CONNECTING) {
          return;
        }
        this.#loop();
      },
      this.#reconnectionTime,
      true,
    );
  }

  #failConnection() {
    if (this.#readyState !== CLOSED) {
      this.#readyState = CLOSED;
      this.dispatchEvent(new Event("error"));
    }
  }

  [SymbolFor("Deno.privateCustomInspect")](inspect, inspectOptions) {
    return inspect(
      createFilteredInspectProxy({
        object: this,
        evaluate: ObjectPrototypeIsPrototypeOf(EventSourcePrototype, this),
        keys: [
          "readyState",
          "url",
          "withCredentials",
          "onopen",
          "onmessage",
          "onerror",
        ],
      }),
      inspectOptions,
    );
  }
}

const EventSourcePrototype = EventSource.prototype;

ObjectDefineProperties(EventSource, {
  CONNECTING: {
    __proto__: null,
    value: 0,
  },
  OPEN: {
    __proto__: null,
    value: 1,
  },
  CLOSED: {
    __proto__: null,
    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,
    },
    { key: "headers", converter: webidl.converters["HeadersInit"] },
  ],
);

return { EventSource, TextLineStream };
})();