feat: add multipart subscription network adapters for Relay and urql

package.json

@@ -127,6 +127,7 @@
     "@types/node-fetch": "2.6.5",
     "@types/react": "18.2.22",
     "@types/react-dom": "18.2.7",
+    "@types/relay-runtime": "^14.1.14",
     "@types/use-sync-external-store": "0.0.4",
     "@typescript-eslint/eslint-plugin": "6.7.2",
     "@typescript-eslint/parser": "6.7.2",

src/utilities/subscriptions/relay.ts

@@ -0,0 +1,60 @@
+import { Observable } from "relay-runtime";
+import type { RequestParameters, GraphQLResponse } from "relay-runtime";
+import {
+  handleError,
+  readMultipartBody,
+} from "../../link/http/parseAndCheckHttpResponse.js";
+import { maybe } from "../index.js";
+import { serializeFetchParameter } from "../../core/index.js";
+
+import type { OperationVariables } from "../../core/index.js";
+import type { Body } from "../../link/http/selectHttpOptionsAndBody.js";
+import { generateOptionsForMultipartSubscription } from "./shared.js";
+import type { CreateMultipartSubscriptionOptions } from "./shared.js";
+
+const backupFetch = maybe(() => fetch);
+
+export function createFetchMultipartSubscription(
+  uri: string,
+  { fetch: preferredFetch, headers }: CreateMultipartSubscriptionOptions = {}
+) {
+  return function fetchMultipartSubscription(
+    operation: RequestParameters,
+    variables: OperationVariables
+  ): Observable<GraphQLResponse> {
+    const body: Body = {
+      operationName: operation.name,
+      variables,
+      query: operation.text || "",
+    };
+    const options = generateOptionsForMultipartSubscription(headers || {});
+
+    return Observable.create((sink) => {
+      try {
+        options.body = serializeFetchParameter(body, "Payload");
+      } catch (parseError) {
+        sink.error(parseError);
+      }
+
+      const currentFetch = preferredFetch || maybe(() => fetch) || backupFetch;
+      const observerNext = sink.next.bind(sink);
+
+      currentFetch!(uri, options)
+        .then((response) => {
+          const ctype = response.headers?.get("content-type");
+
+          if (ctype !== null && /^multipart\/mixed/i.test(ctype)) {
+            return readMultipartBody(response, observerNext);
+          }
+
+          sink.error(new Error("Expected multipart response"));
+        })
+        .then(() => {
+          sink.complete();
+        })
+        .catch((err: any) => {
+          handleError(err, sink);
+        });
+    });
+  };
+}

src/utilities/subscriptions/shared.ts

@@ -0,0 +1,21 @@
+import { fallbackHttpConfig } from "../../link/http/selectHttpOptionsAndBody.js";
+
+export type CreateMultipartSubscriptionOptions = {
+  fetch?: WindowOrWorkerGlobalScope["fetch"];
+  headers?: Record<string, string>;
+};
+
+export function generateOptionsForMultipartSubscription(
+  headers: Record<string, string>
+) {
+  const options: { headers: Record<string, any>; body?: string } = {
+    ...fallbackHttpConfig.options,
+    headers: {
+      ...(headers || {}),
+      ...fallbackHttpConfig.headers,
+      accept:
+        "multipart/mixed;boundary=graphql;subscriptionSpec=1.0,application/json",
+    },
+  };
+  return options;
+}

src/utilities/subscriptions/urql.ts

@@ -0,0 +1,56 @@
+import { Observable } from "../index.js";
+import {
+  handleError,
+  readMultipartBody,
+} from "../../link/http/parseAndCheckHttpResponse.js";
+import { maybe } from "../index.js";
+import { serializeFetchParameter } from "../../core/index.js";
+import type { Body } from "../../link/http/selectHttpOptionsAndBody.js";
+import { generateOptionsForMultipartSubscription } from "./shared.js";
+import type { CreateMultipartSubscriptionOptions } from "./shared.js";
+
+const backupFetch = maybe(() => fetch);
+
+export function createFetchMultipartSubscription(
+  uri: string,
+  { fetch: preferredFetch, headers }: CreateMultipartSubscriptionOptions = {}
+) {
+  return function multipartSubscriptionForwarder({
+    query,
+    variables,
+  }: {
+    query?: string;
+    variables: undefined | Record<string, any>;
+  }) {
+    const body: Body = { variables, query };
+    const options = generateOptionsForMultipartSubscription(headers || {});
+
+    return new Observable((observer) => {
+      try {
+        options.body = serializeFetchParameter(body, "Payload");
+      } catch (parseError) {
+        observer.error(parseError);
+      }
+
+      const currentFetch = preferredFetch || maybe(() => fetch) || backupFetch;
+      const observerNext = observer.next.bind(observer);
+
+      currentFetch!(uri, options)
+        .then((response) => {
+          const ctype = response.headers?.get("content-type");
+
+          if (ctype !== null && /^multipart\/mixed/i.test(ctype)) {
+            return readMultipartBody(response, observerNext);
+          }
+
+          observer.error(new Error("Expected multipart response"));
+        })
+        .then(() => {
+          observer.complete();
+        })
+        .catch((err: any) => {
+          handleError(err, observer);
+        });
+    });
+  };
+}