Switch to ReadableStream (by gpt)

Author: MaxLeiter

lib/stream.js

@@ -1,136 +1,100 @@
-// Attempt to use the native ReadbleStream if available, otherwise check for the `readable-stream` package or the built-in Node.js `stream` module.
-let Readable;
-
-if (typeof ReadableStream === 'undefined') {
-  try {
-    Readable = require("readable-stream").Readable || require("stream").Readable;
-  } catch (e) {
-    Readable = null;
-  }
-} else {
-  Readable = ReadableStream;
-}
-
-/**
- * A server-sent event.
- */
 class ServerSentEvent {
-  /**
-   * Create a new server-sent event.
-   *
-   * @param {string} event The event name.
-   * @param {string} data The event data.
-   * @param {string} id The event ID.
-   * @param {number} retry The retry time.
-   */
   constructor(event, data, id, retry) {
     this.event = event;
     this.data = data;
     this.id = id;
     this.retry = retry;
   }
 
-  /**
-   * Convert the event to a string.
-   */
   toString() {
-    if (this.event === "output") {
-      return this.data;
+    let result = '';
+    if (this.id) {
+      result += `id: ${this.id}\n`;
     }
-
-    return "";
+    if (this.event) {
+      result += `event: ${this.event}\n`;
+    }
+    if (this.retry) {
+      result += `retry: ${this.retry}\n`;
+    }
+    if (this.data) {
+      result += `data: ${this.data}\n`;
+    }
+    result += '\n';
+    return result;
   }
 }
 
-/**
- * A stream of server-sent events.
- */
-class Stream extends Readable {
-  /**
-   * Create a new stream of server-sent events.
-   *
-   * @param {string} url The URL to connect to.
-   * @param {object} options The fetch options.
-   */
+class Stream {
   constructor(url, options) {
-    if (!Readable) {
-      throw new Error(
-        "Readable streams are not supported. Please use Node.js 18 or later, or install the readable-stream package."
-      );
-    }
-
-    super();
     this.url = url;
     this.options = options;
+    this.readableStream = new ReadableStream({
+      start: async (controller) => {
+        const response = await fetch(this.url, {
+          ...this.options,
+          headers: {
+            Accept: 'text/event-stream',
+          },
+        });
+        const reader = response.body.getReader();
+        let decoder = new TextDecoder();
+        let eventBuffer = '';
+
+        const processChunk = (chunk) => {
+          eventBuffer += decoder.decode(chunk, {stream: true});
+          let eolIndex;
+          while ((eolIndex = eventBuffer.indexOf('\n')) >= 0) {
+            const line = eventBuffer.slice(0, eolIndex).trim();
+            eventBuffer = eventBuffer.slice(eolIndex + 1);
+            if (line === '') {
+              // End of an event
+              const event = this.parseEvent(eventBuffer);
+              controller.enqueue(event);
+              eventBuffer = '';
+            } else {
+              // Accumulate data
+              eventBuffer += line + '\n';
+            }
+          }
+        };
 
-    this.event = null;
-    this.data = [];
-    this.lastEventId = null;
-    this.retry = null;
-  }
+        const push = async () => {
+          const {done, value} = await reader.read();
+          if (done) {
+            controller.close();
+            return;
+          }
+          processChunk(value);
+          push();
+        };
 
-  decode(line) {
-    if (!line) {
-      if (!this.event && !this.data.length && !this.lastEventId) {
-        return null;
+        push();
       }
-
-      const sse = new ServerSentEvent(
-        this.event,
-        this.data.join("\n"),
-        this.lastEventId
-      );
-
-      this.event = null;
-      this.data = [];
-      this.retry = null;
-
-      return sse;
-    }
-
-    if (line.startsWith(":")) {
-      return null;
-    }
-
-    const [field, value] = line.split(": ");
-    if (field === "event") {
-      this.event = value;
-    } else if (field === "data") {
-      this.data.push(value);
-    } else if (field === "id") {
-      this.lastEventId = value;
-    }
-
-    return null;
-  }
-
-  async *[Symbol.asyncIterator]() {
-    const response = await fetch(this.url, {
-      ...this.options,
-      headers: {
-        Accept: "text/event-stream",
-      },
     });
+  }
 
-    for await (const chunk of response.body) {
-      const decoder = new TextDecoder("utf-8");
-      const text = decoder.decode(chunk);
-      const lines = text.split("\n");
-      for (const line of lines) {
-        const sse = this.decode(line);
-        if (sse) {
-          if (sse.event === "error") {
-            throw new Error(sse.data);
-          }
-
-          yield sse;
-
-          if (sse.event === "done") {
-            return;
-          }
-        }
+  parseEvent(rawData) {
+    const lines = rawData.trim().split('\n');
+    let event = 'message', data = '', id = null, retry = null;
+    for (const line of lines) {
+      const [fieldName, value] = line.split(/:(.*)/, 2);
+      switch (fieldName) {
+        case 'event':
+          event = value.trim();
+          break;
+        case 'data':
+          data += value.trim() + '\n';
+          break;
+        case 'id':
+          id = value.trim();
+          break;
+        case 'retry':
+          retry = parseInt(value.trim(), 10);
+          break;
       }
     }
+    return new ServerSentEvent(event, data.trim(), id, retry);
   }
 }