Normalize the stream source to an async iterable before passing to the writers

Author: ericallam

packages/core/src/v3/realtimeStreams/manager.ts

@@ -1,6 +1,7 @@
 import {
   AsyncIterableStream,
   createAsyncIterableStreamFromAsyncIterable,
+  ensureAsyncIterable,
 } from "../streams/asyncIterableStream.js";
 import {
   RealtimeAppendStreamOptions,
@@ -30,6 +31,9 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
     source: AsyncIterable<T> | ReadableStream<T>,
     options?: RealtimeAppendStreamOptions
   ): Promise<RealtimeStreamInstance<T>> {
+    // Normalize ReadableStream to AsyncIterable
+    const asyncIterableSource = ensureAsyncIterable(source);
+
     const runId = getRunIdForOptions(options);
 
     if (!runId) {
@@ -52,7 +56,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
         ? new MetadataStream({
             key,
             runId,
-            source,
+            source: asyncIterableSource,
             baseUrl: this.baseUrl,
             headers: this.apiClient.getHeaders(),
             signal: options?.signal,
@@ -63,7 +67,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
             basin: parsedResponse.basin,
             stream: key,
             accessToken: parsedResponse.accessToken,
-            source,
+            source: asyncIterableSource,
             signal: options?.signal,
             limiter: (await import("p-limit")).default,
             debug: this.debug,

packages/core/src/v3/streams/asyncIterableStream.ts

@@ -103,3 +103,33 @@ export function createAsyncIterableStreamFromAsyncGenerator<T>(
 ): AsyncIterableStream<T> {
   return createAsyncIterableStreamFromAsyncIterable(asyncGenerator, transformer, signal);
 }
+
+export function ensureAsyncIterable<T>(
+  input: AsyncIterable<T> | ReadableStream<T>
+): AsyncIterable<T> {
+  // If it's already an AsyncIterable, return it as-is
+  if (Symbol.asyncIterator in input) {
+    return input as AsyncIterable<T>;
+  }
+
+  // Convert ReadableStream to AsyncIterable
+  const readableStream = input as ReadableStream<T>;
+  return {
+    async *[Symbol.asyncIterator]() {
+      const reader = readableStream.getReader();
+      try {
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) {
+            break;
+          }
+          if (value !== undefined) {
+            yield value;
+          }
+        }
+      } finally {
+        reader.releaseLock();
+      }
+    },
+  };
+}