implement multipart streaming for PreloadQuery

integration-test/nextjs/src/app/rsc/dynamic/PreloadQuery/queryRef-useReadQuery/page.tsx

@@ -6,14 +6,14 @@ export const dynamic = "force-dynamic";
 import { PreloadQuery } from "../../../client";
 import { Suspense } from "react";
 
-export default function Page({ searchParams }: { searchParams?: any }) {
+export default async function Page({ searchParams }: { searchParams?: any }) {
   return (
     <ApolloWrapper>
       <PreloadQuery
         query={QUERY}
         context={{
           delay: 1000,
-          error: searchParams?.errorIn || undefined,
+          error: (await searchParams)?.errorIn || undefined,
         }}
       >
         {(queryRef) => (

integration-test/nextjs/src/app/rsc/dynamic/PreloadQuery/useSuspenseQuery/page.tsx

@@ -6,14 +6,14 @@ export const dynamic = "force-dynamic";
 import { PreloadQuery } from "../../../client";
 import { Suspense } from "react";
 
-export default function Page({ searchParams }: { searchParams?: any }) {
+export default async function Page({ searchParams }: { searchParams?: any }) {
   return (
     <ApolloWrapper>
       <PreloadQuery
         query={QUERY}
         context={{
           delay: 1000,
-          error: searchParams?.errorIn || undefined,
+          error: (await searchParams)?.errorIn || undefined,
         }}
       >
         <Suspense fallback={<>loading</>}>

integration-test/nextjs/src/shared/errorLink.tsx

@@ -1,5 +1,5 @@
 import { ApolloLink, Observable } from "@apollo/client";
-import { GraphQLError } from "graphql";
+import { GraphQLError, GraphQLFormattedError } from "graphql";
 import * as entryPoint from "@apollo/client-react-streaming";
 
 declare module "@apollo/client" {
@@ -22,7 +22,11 @@ export const errorLink = new ApolloLink((operation, forward) => {
     return new Observable((subscriber) => {
       subscriber.next({
         data: null,
-        errors: [new GraphQLError("Simulated error")],
+        errors: [
+          {
+            message: "Simulated error",
+          } satisfies GraphQLFormattedError as GraphQLError,
+        ],
       });
     });
   }

integration-test/yarn.lock

@@ -32,7 +32,7 @@ __metadata:
   linkType: hard
 
 "@apollo/client-react-streaming@exec:./shared/build-client-react-streaming.cjs::locator=%40integration-test%2Froot%40workspace%3A.":
-  version: 0.11.4
+  version: 0.11.5
   resolution: "@apollo/client-react-streaming@exec:./shared/build-client-react-streaming.cjs#./shared/build-client-react-streaming.cjs::hash=48b117&locator=%40integration-test%2Froot%40workspace%3A."
   dependencies:
     ts-invariant: "npm:^0.10.3"
@@ -81,10 +81,10 @@ __metadata:
   linkType: hard
 
 "@apollo/experimental-nextjs-app-support@exec:./shared/build-experimental-nextjs-app-support.cjs::locator=%40integration-test%2Froot%40workspace%3A.":
-  version: 0.11.4
+  version: 0.11.5
   resolution: "@apollo/experimental-nextjs-app-support@exec:./shared/build-experimental-nextjs-app-support.cjs#./shared/build-experimental-nextjs-app-support.cjs::hash=fd83cc&locator=%40integration-test%2Froot%40workspace%3A."
   dependencies:
-    "@apollo/client-react-streaming": "npm:0.11.4"
+    "@apollo/client-react-streaming": "npm:0.11.5"
   peerDependencies:
     "@apollo/client": ^3.10.4
     next: ^13.4.1 || ^14.0.0 || ^15.0.0-rc.0

packages/client-react-streaming/package-shape.json

@@ -7,6 +7,10 @@
       "RemoveMultipartDirectivesLink",
       "SSRMultipartLink",
       "registerApolloClient",
+      "ReadFromReadableStreamLink",
+      "TeeToReadableStreamLink",
+      "readFromReadableStream",
+      "teeToReadableStream",
       "built_for_rsc"
     ],
     "browser": [
@@ -17,7 +21,12 @@
       "RemoveMultipartDirectivesLink",
       "SSRMultipartLink",
       "WrapApolloProvider",
+      "skipDataTransport",
       "resetApolloSingletons",
+      "ReadFromReadableStreamLink",
+      "TeeToReadableStreamLink",
+      "readFromReadableStream",
+      "teeToReadableStream",
       "built_for_browser"
     ],
     "node": [
@@ -28,7 +37,12 @@
       "RemoveMultipartDirectivesLink",
       "SSRMultipartLink",
       "WrapApolloProvider",
+      "skipDataTransport",
       "resetApolloSingletons",
+      "ReadFromReadableStreamLink",
+      "TeeToReadableStreamLink",
+      "readFromReadableStream",
+      "teeToReadableStream",
       "built_for_ssr"
     ],
     "edge-light,worker,browser": [
@@ -39,7 +53,12 @@
       "RemoveMultipartDirectivesLink",
       "SSRMultipartLink",
       "WrapApolloProvider",
+      "skipDataTransport",
       "resetApolloSingletons",
+      "ReadFromReadableStreamLink",
+      "TeeToReadableStreamLink",
+      "readFromReadableStream",
+      "teeToReadableStream",
       "built_for_ssr"
     ]
   },

packages/client-react-streaming/package.json

@@ -78,18 +78,18 @@
     },
     "./stream-utils": {
       "require": {
-        "types": "./dist/stream-utils.node.d.cts",
-        "edge-light": "./dist/stream-utils.node.cjs",
-        "react-server": "./dist/empty.cjs",
-        "browser": "./dist/empty.cjs",
-        "node": "./dist/stream-utils.node.cjs"
+        "types": "./dist/stream-utils.ssr.d.cts",
+        "react-server": "./dist/stream-utils.cjs",
+        "edge-light": "./dist/stream-utils.ssr.cjs",
+        "browser": "./dist/stream-utils.cjs",
+        "node": "./dist/stream-utils.ssr.cjs"
       },
       "import": {
-        "types": "./dist/stream-utils.node.d.ts",
-        "edge-light": "./dist/stream-utils.node.js",
-        "react-server": "./dist/empty.js",
-        "browser": "./dist/empty.js",
-        "node": "./dist/stream-utils.node.js"
+        "types": "./dist/stream-utils.rsc.d.ts",
+        "react-server": "./dist/stream-utils.js",
+        "edge-light": "./dist/stream-utils.ssr.js",
+        "browser": "./dist/stream-utils.js",
+        "node": "./dist/stream-utils.ssr.js"
       }
     },
     "./package.json": "./package.json"

packages/client-react-streaming/src/DataTransportAbstraction/WrappedApolloClient.tsx

@@ -6,6 +6,7 @@ import type {
   FetchResult,
   DocumentNode,
   NormalizedCacheObject,
+  ApolloLink,
 } from "@apollo/client/index.js";
 import {
   ApolloClient as OrigApolloClient,
@@ -28,6 +29,10 @@ import type {
 import { bundle, sourceSymbol } from "../bundleInfo.js";
 import { serializeOptions, deserializeOptions } from "./transportedOptions.js";
 import { assertInstance } from "../assertInstance.js";
+import {
+  ReadFromReadableStreamLink,
+  TeeToReadableStreamLink,
+} from "../ReadableStreamLink.js";
 
 function getQueryManager(
   client: OrigApolloClient<unknown>
@@ -298,9 +303,37 @@ export class ApolloClientClientBaseImpl extends ApolloClientBase {
   };
 }
 
+const skipDataTransportKey = Symbol.for("apollo.dataTransport.skip");
+interface InternalContext {
+  [skipDataTransportKey]?: boolean;
+}
+
+/**
+ * Apply to a context to prevent this operation from being transported over the SSR data transport mechanism.
+ * @param readableStream
+ * @param context
+ * @returns
+ */
+export function skipDataTransport<T extends Record<string, any>>(
+  context: T
+): T & InternalContext {
+  return Object.assign(context, {
+    [skipDataTransportKey]: true,
+  });
+}
+
 class ApolloClientSSRImpl extends ApolloClientClientBaseImpl {
   private forwardedQueries = new (getTrieConstructor(this))();
 
+  constructor(options: WrappedApolloClientOptions) {
+    super(options);
+    this.setLink(this.link);
+  }
+
+  setLink(newLink: ApolloLink) {
+    super.setLink.call(this, ReadFromReadableStreamLink.concat(newLink));
+  }
+
   watchQueryQueue = createBackpressuredCallback<{
     event: Extract<QueryEvent, { type: "started" }>;
     observable: Observable<Exclude<QueryEvent, { type: "started" }>>;
@@ -315,6 +348,9 @@ class ApolloClientSSRImpl extends ApolloClientClientBaseImpl {
     if (
       options.fetchPolicy !== "cache-only" &&
       options.fetchPolicy !== "standby" &&
+      !(options.context as InternalContext | undefined)?.[
+        skipDataTransportKey
+      ] &&
       !this.forwardedQueries.peekArray(cacheKeyArr)
     ) {
       // don't transport the same query over twice
@@ -376,14 +412,34 @@ class ApolloClientSSRImpl extends ApolloClientClientBaseImpl {
   }
 }
 
-export class ApolloClientBrowserImpl extends ApolloClientClientBaseImpl {}
+export class ApolloClientBrowserImpl extends ApolloClientClientBaseImpl {
+  constructor(options: WrappedApolloClientOptions) {
+    super(options);
+    this.setLink(this.link);
+  }
+
+  setLink(newLink: ApolloLink) {
+    super.setLink.call(this, ReadFromReadableStreamLink.concat(newLink));
+  }
+}
+
+export class ApolloClientRSCImpl extends ApolloClientBase {
+  constructor(options: WrappedApolloClientOptions) {
+    super(options);
+    this.setLink(this.link);
+  }
+
+  setLink(newLink: ApolloLink) {
+    super.setLink.call(this, TeeToReadableStreamLink.concat(newLink));
+  }
+}
 
 const ApolloClientImplementation =
   /*#__PURE__*/ process.env.REACT_ENV === "ssr"
     ? ApolloClientSSRImpl
     : process.env.REACT_ENV === "browser"
       ? ApolloClientBrowserImpl
-      : ApolloClientBase;
+      : ApolloClientRSCImpl;
 
 /**
  * A version of `ApolloClient` to be used with streaming SSR or in React Server Components.

packages/client-react-streaming/src/DataTransportAbstraction/index.ts

@@ -1,5 +1,5 @@
 export { InMemoryCache } from "./WrappedInMemoryCache.js";
-export { ApolloClient } from "./WrappedApolloClient.js";
+export { ApolloClient, skipDataTransport } from "./WrappedApolloClient.js";
 
 export { resetApolloSingletons } from "./testHelpers.js";
 

packages/client-react-streaming/src/PreloadQuery.tsx

@@ -1,30 +1,26 @@
 import { SimulatePreloadedQuery } from "./index.cc.js";
 import type {
   ApolloClient,
+  DocumentNode,
   OperationVariables,
-  QueryOptions,
+  TypedDocumentNode,
 } from "@apollo/client";
 import type { ReactNode } from "react";
 import React from "react";
-import { serializeOptions } from "./DataTransportAbstraction/transportedOptions.js";
-import type { TransportedQueryRef } from "./transportedQueryRef.js";
-import { createTransportedQueryRef } from "./transportedQueryRef.js";
-import type { ProgressEvent } from "./DataTransportAbstraction/DataTransportAbstraction.js";
-
-export type RestrictedPreloadOptions = {
-  fetchPolicy?: "cache-first";
-  returnPartialData?: false;
-  nextFetchPolicy?: undefined;
-  pollInterval?: undefined;
-};
-
-export type PreloadQueryOptions<TVariables, TData> = QueryOptions<
-  TVariables,
-  TData
-> &
-  RestrictedPreloadOptions;
-
-export function PreloadQuery<TData, TVariables extends OperationVariables>({
+import type {
+  PreloadTransportedQueryOptions,
+  TransportedQueryRef,
+} from "./transportedQueryRef.js";
+import { createTransportedQueryPreloader } from "./transportedQueryRef.js";
+
+export type PreloadQueryOptions<TVariables, TData> =
+  PreloadTransportedQueryOptions<TVariables, TData> & {
+    query: DocumentNode | TypedDocumentNode<TData, TVariables>;
+  };
+export async function PreloadQuery<
+  TData,
+  TVariables extends OperationVariables,
+>({
   getClient,
   children,
   ...options
@@ -35,50 +31,14 @@ export function PreloadQuery<TData, TVariables extends OperationVariables>({
     | ((
         queryRef: TransportedQueryRef<NoInfer<TData>, NoInfer<TVariables>>
       ) => ReactNode);
-}): React.ReactElement {
-  const preloadOptions = {
-    ...options,
-    fetchPolicy: "cache-first" as const,
-    returnPartialData: false,
-    pollInterval: undefined,
-    nextFetchPolicy: undefined,
-  } satisfies RestrictedPreloadOptions;
-
-  const transportedOptions = sanitizeForTransport(
-    serializeOptions(preloadOptions)
-  );
-
-  const resultPromise = Promise.resolve(getClient())
-    .then((client) => client.query<TData, TVariables>(preloadOptions))
-    .then<Array<Omit<ProgressEvent, "id">>, Array<Omit<ProgressEvent, "id">>>(
-      (result) => [
-        { type: "data", result: sanitizeForTransport(result) },
-        { type: "complete" },
-      ],
-      () => [{ type: "error" }]
-    );
-
-  const queryKey = crypto.randomUUID();
+}): Promise<React.ReactElement> {
+  const preloader = createTransportedQueryPreloader(await getClient());
+  const { query, ...transportedOptions } = options;
+  const queryRef = preloader(query, transportedOptions);
 
   return (
-    <SimulatePreloadedQuery<TData>
-      options={transportedOptions}
-      result={resultPromise}
-      queryKey={typeof children === "function" ? queryKey : undefined}
-    >
-      {typeof children === "function"
-        ? children(
-            createTransportedQueryRef<TData, TVariables>(
-              transportedOptions,
-              queryKey,
-              resultPromise
-            )
-          )
-        : children}
+    <SimulatePreloadedQuery<TData> queryRef={queryRef}>
+      {typeof children === "function" ? children(queryRef) : children}
     </SimulatePreloadedQuery>
   );
 }
-
-function sanitizeForTransport<T>(value: T) {
-  return JSON.parse(JSON.stringify(value)) as T;
-}