2939

examples/next-openai/app/movies/action.tsx

@@ -0,0 +1,41 @@
+'use server';
+
+import { createStreamableUI } from 'ai/rsc';
+
+import React from 'react';
+
+const MovieItem = () => {
+  return (
+    <div
+      className="block p-4 text-white bg-blue-500"
+      style={{
+        animation: 'fadeIn 0.5s ease-in-out',
+      }}
+    >
+      <h2 className="text-xl font-bold">Movie Title</h2>
+      <p className="mt-2">
+        This is a placeholder for movie description. It will be replaced with
+        actual content later.
+      </p>
+    </div>
+  );
+};
+
+export async function getMovies() {
+  const movieUi = createStreamableUI(null);
+
+  setTimeout(() => {
+    movieUi.update(<MovieItem />);
+  }, 1000);
+
+  setTimeout(() => {
+    movieUi.append(<MovieItem />);
+  }, 2000);
+
+  setTimeout(() => {
+    movieUi.append(<MovieItem />);
+    movieUi.done();
+  }, 3000);
+
+  return movieUi.value;
+}

examples/next-openai/app/movies/page.tsx

@@ -0,0 +1,32 @@
+'use client';
+
+import { useState } from 'react';
+import { getMovies } from './action';
+
+export default function Page() {
+  const [movies, setMovies] = useState<React.ReactNode | null>(null);
+
+  return (
+    <div>
+      <style jsx>{`
+        @keyframes fadeIn {
+          from {
+            opacity: 0;
+          }
+          to {
+            opacity: 1;
+          }
+        }
+      `}</style>
+      <button
+        onClick={async () => {
+          const moviesUI = await getMovies();
+          setMovies(moviesUI);
+        }}
+      >
+        What&apos;s the weather?
+      </button>
+      {movies}
+    </div>
+  );
+}

packages/ai/rsc/streamable-ui/create-streamable-ui.tsx → packages/ai/rsc/create-streamable-ui.tsx

@@ -1,6 +1,6 @@
-import { HANGING_STREAM_WARNING_TIME_MS } from '../../util/constants';
-import { createResolvablePromise } from '../../util/create-resolvable-promise';
-import { createSuspendedChunk } from './create-suspended-chunk';
+import { HANGING_STREAM_WARNING_TIME_MS } from './../util/constants';
+import { InternalStreamableUIClient } from './rsc-shared.mjs';
+import { createStreamableValue } from './streamable-value/create-streamable-value';
 
 // It's necessary to define the type manually here, otherwise TypeScript compiler
 // will not be able to infer the correct return type as it's circular.
@@ -53,9 +53,8 @@ type StreamableUIWrapper = {
  * On the client side, it can be rendered as a normal React node.
  */
 function createStreamableUI(initialValue?: React.ReactNode) {
-  let currentValue = initialValue;
+  const innerStreamable = createStreamableValue<React.ReactNode>(initialValue);
   let closed = false;
-  let { row, resolve, reject } = createSuspendedChunk(initialValue);
 
   function assertStream(method: string) {
     if (closed) {
@@ -79,37 +78,19 @@ function createStreamableUI(initialValue?: React.ReactNode) {
   warnUnclosedStream();
 
   const streamable: StreamableUIWrapper = {
-    value: row,
+    value: <InternalStreamableUIClient s={innerStreamable.value} />,
     update(value: React.ReactNode) {
       assertStream('.update()');
 
-      // There is no need to update the value if it's referentially equal.
-      if (value === currentValue) {
-        warnUnclosedStream();
-        return streamable;
-      }
-
-      const resolvable = createResolvablePromise();
-      currentValue = value;
-
-      resolve({ value: currentValue, done: false, next: resolvable.promise });
-      resolve = resolvable.resolve;
-      reject = resolvable.reject;
-
+      innerStreamable.update(value);
       warnUnclosedStream();
 
       return streamable;
     },
     append(value: React.ReactNode) {
       assertStream('.append()');
 
-      const resolvable = createResolvablePromise();
-      currentValue = value;
-
-      resolve({ value, done: false, append: true, next: resolvable.promise });
-      resolve = resolvable.resolve;
-      reject = resolvable.reject;
-
+      innerStreamable.append(value);
       warnUnclosedStream();
 
       return streamable;
@@ -121,7 +102,7 @@ function createStreamableUI(initialValue?: React.ReactNode) {
         clearTimeout(warningTimeout);
       }
       closed = true;
-      reject(error);
+      innerStreamable.error(error);
 
       return streamable;
     },
@@ -133,11 +114,11 @@ function createStreamableUI(initialValue?: React.ReactNode) {
       }
       closed = true;
       if (args.length) {
-        resolve({ value: args[0], done: true });
+        innerStreamable.done(args[0]);
         return streamable;
       }
-      resolve({ value: currentValue, done: true });
 
+      innerStreamable.done();
       return streamable;
     },
   };

packages/ai/rsc/rsc-server.ts

@@ -1,5 +1,5 @@
 export { getAIState, getMutableAIState } from './ai-state';
 export { createAI } from './provider';
 export { render, streamUI } from './stream-ui';
-export { createStreamableUI } from './streamable-ui/create-streamable-ui';
+export { createStreamableUI } from './create-streamable-ui';
 export { createStreamableValue } from './streamable-value/create-streamable-value';

packages/ai/rsc/rsc-shared.mts

@@ -8,4 +8,5 @@ export {
   useActions,
   useSyncUIState,
   InternalAIProvider,
+  InternalStreamableUIClient,
 } from './shared-client';

packages/ai/rsc/shared-client/index.ts

@@ -9,3 +9,4 @@ export {
   useSyncUIState,
   useUIState,
 } from './context';
+export { InternalStreamableUIClient } from './internal-streamable-ui-client';

packages/ai/rsc/shared-client/internal-streamable-ui-client.tsx

@@ -0,0 +1,56 @@
+'use client';
+
+import { useEffect, useState } from 'react';
+import { readStreamableValue } from '../streamable-value/read-streamable-value';
+import { StreamableValue } from '../streamable-value/streamable-value';
+
+export function InternalStreamableUIClient<T>({
+  s,
+}: {
+  s: StreamableValue<T>;
+}) {
+  // Set the value to the initial value of the streamable, if it has one.
+  const [value, setValue] = useState<T | undefined>(s.curr);
+
+  // Error state for the streamable. It might be errored initially and we want
+  // to error out as soon as possible.
+  const [error, setError] = useState<Error | undefined>(s.error);
+
+  useEffect(() => {
+    let canceled = false;
+    setError(undefined);
+
+    (async () => {
+      try {
+        // Read the streamable value and update the state with the new value.
+        for await (const v of readStreamableValue(s)) {
+          if (canceled) {
+            break;
+          }
+
+          setValue(v);
+        }
+      } catch (e) {
+        if (canceled) {
+          return;
+        }
+
+        setError(e as Error);
+      }
+    })();
+
+    return () => {
+      // If the component is unmounted, we want to cancel the stream.
+      canceled = true;
+    };
+  }, [s]);
+
+  // This ensures that errors from the streamable UI are thrown during the
+  // render phase, so that they can be caught by error boundary components.
+  // This is necessary for React's declarative model.
+  if (error) {
+    throw error;
+  }
+
+  return value;
+}

packages/ai/rsc/stream-ui/render.ts

@@ -5,7 +5,7 @@ import zodToJsonSchema from 'zod-to-json-schema';
 import { OpenAIStream } from '../../streams';
 import { consumeStream } from '../../util/consume-stream';
 import { createResolvablePromise } from '../../util/create-resolvable-promise';
-import { createStreamableUI } from '../streamable-ui/create-streamable-ui';
+import { createStreamableUI } from '../create-streamable-ui';
 
 type Streamable = ReactNode | Promise<ReactNode>;
 type Renderer<T> = (

packages/ai/rsc/stream-ui/stream-ui.tsx

@@ -24,7 +24,7 @@ import { createResolvablePromise } from '../../util/create-resolvable-promise';
 import { isAsyncGenerator } from '../../util/is-async-generator';
 import { isGenerator } from '../../util/is-generator';
 import { retryWithExponentialBackoff } from '../../util/retry-with-exponential-backoff';
-import { createStreamableUI } from '../streamable-ui/create-streamable-ui';
+import { createStreamableUI } from '../create-streamable-ui';
 
 type Streamable = ReactNode | Promise<ReactNode>;
 

packages/ai/rsc/streamable-value/create-streamable-value.ts → packages/ai/rsc/streamable-value/create-streamable-value.tsx

@@ -1,3 +1,4 @@
+import { isValidElement, ReactElement } from 'react';
 import { HANGING_STREAM_WARNING_TIME_MS } from '../../util/constants';
 import { createResolvablePromise } from '../../util/create-resolvable-promise';
 import {
@@ -223,31 +224,39 @@ function createStreamableValueImpl<T = any, E = any>(initialValue?: T) {
     append(value: T) {
       assertStream('.append()');
 
-      if (
-        typeof currentValue !== 'string' &&
-        typeof currentValue !== 'undefined'
-      ) {
-        throw new Error(
-          `.append(): The current value is not a string. Received: ${typeof currentValue}`,
+      if (typeof currentValue === 'undefined') {
+        currentPatchValue = undefined;
+        currentValue = value;
+      } else if (typeof currentValue === 'string') {
+        if (typeof value === 'string') {
+          currentPatchValue = [0, value];
+          (currentValue as string) = currentValue + value;
+        } else {
+          currentPatchValue = [1, value as ReactElement];
+          (currentValue as unknown as ReactElement) = (
+            <>
+              {currentValue}
+              {value}
+            </>
+          );
+        }
+      } else if (isValidElement(currentValue)) {
+        currentPatchValue = [1, value as ReactElement];
+        (currentValue as ReactElement) = (
+          <>
+            {currentValue}
+            {value}
+          </>
         );
-      }
-      if (typeof value !== 'string') {
+      } else {
         throw new Error(
-          `.append(): The value is not a string. Received: ${typeof value}`,
+          `.append(): The current value doesn't support appending data. Type: ${typeof currentValue}`,
         );
       }
 
       const resolvePrevious = resolvable.resolve;
       resolvable = createResolvablePromise();
 
-      if (typeof currentValue === 'string') {
-        currentPatchValue = [0, value];
-        (currentValue as string) = currentValue + value;
-      } else {
-        currentPatchValue = undefined;
-        currentValue = value;
-      }
-
       currentPromise = resolvable.promise;
       resolvePrevious(createWrapped());