feat: XStream 返回流同时支持异步迭代

components/x-stream/index.ts

@@ -137,11 +137,13 @@ export interface XStreamOptions<Output> {
   transformStream?: TransformStream<string, Output>;
 }
 
+type XReadableStream<R = SSEOutput> = ReadableStream<R> & AsyncGenerator<R>;
+
 /**
  * @description Transform Uint8Array binary stream to {@link SSEOutput} by default
  * @warning The `XStream` only support the `utf-8` encoding. More encoding support maybe in the future.
  */
-async function* XStream<Output = SSEOutput>(options: XStreamOptions<Output>) {
+function XStream<Output = SSEOutput>(options: XStreamOptions<Output>) {
   const { readableStream, transformStream } = options;
 
   if (!(readableStream instanceof ReadableStream)) {
@@ -151,33 +153,38 @@ async function* XStream<Output = SSEOutput>(options: XStreamOptions<Output>) {
   // Default encoding is `utf-8`
   const decoderStream = new TextDecoderStream();
 
-  const stream = transformStream
-    ? /**
-       * Uint8Array binary -> string -> Output
-       */
-      readableStream
-        .pipeThrough(decoderStream)
-        .pipeThrough(transformStream)
-    : /**
-       * Uint8Array binary -> string -> SSE part string -> Default Output {@link SSEOutput}
-       */
-      readableStream
-        .pipeThrough(decoderStream)
-        .pipeThrough(splitStream())
-        .pipeThrough(splitPart());
-
-  const reader = stream.getReader() as ReadableStreamDefaultReader<Output>;
-
-  while (reader instanceof ReadableStreamDefaultReader) {
-    const { value, done } = await reader.read();
-
-    if (done) break;
-
-    if (!value) continue;
-
-    // Transformed data through all transform pipes
-    yield value;
+  const stream = (
+    transformStream
+      ? /**
+         * Uint8Array binary -> string -> Output
+         */
+        readableStream.pipeThrough(decoderStream).pipeThrough(transformStream)
+      : /**
+         * Uint8Array binary -> string -> SSE part string -> Default Output {@link SSEOutput}
+         */
+        readableStream
+          .pipeThrough(decoderStream)
+          .pipeThrough(splitStream())
+          .pipeThrough(splitPart())
+  ) as XReadableStream<Output>;
+
+  /** support async iterator */
+  stream[Symbol.asyncIterator] = async function* () {
+    const reader = this.getReader();
+
+    while (true) {
+      const { done, value } = await reader.read();
+
+      if (done) break;
+
+      if (!value) continue;
+
+      // Transformed data through all transform pipes
+      yield value;
+    }
   }
+
+  return stream;
 }
 
 export default XStream;