feat: Multipart Subscriptions (#10592)

.changeset/nervous-tomatoes-act.md

@@ -0,0 +1,5 @@
+---
+"@apollo/client": patch
+---
+
+Adds support for multipart subscriptions in `HttpLink`.

config/bundlesize.ts

@@ -3,7 +3,7 @@ import { join } from "path";
 import { gzipSync } from "zlib";
 import bytes from "bytes";
 
-const gzipBundleByteLengthLimit = bytes("32.75KB");
+const gzipBundleByteLengthLimit = bytes("33.02KB");
 const minFile = join("dist", "apollo-client.min.cjs");
 const minPath = join(__dirname, "..", minFile);
 const gzipByteLen = gzipSync(readFileSync(minPath)).byteLength;

config/jest.config.js

@@ -30,15 +30,14 @@ const react18TestFileIgnoreList = [
   // to avoid running them twice with both react versions
   // since they do not import react
   ignoreTSFiles,
-  // failing hoc tests (8)
+  // failing hoc tests (7)
   'src/react/hoc/__tests__/mutations/queries.test.tsx',
   'src/react/hoc/__tests__/mutations/recycled-queries.test.tsx',
   'src/react/hoc/__tests__/queries/errors.test.tsx',
   'src/react/hoc/__tests__/queries/lifecycle.test.tsx',
   'src/react/hoc/__tests__/queries/loading.test.tsx',
   'src/react/hoc/__tests__/queries/observableQuery.test.tsx',
   'src/react/hoc/__tests__/queries/skip.test.tsx',
-  'src/react/hoc/__tests__/subscriptions/subscriptions.test.tsx',
   // failing components tests (1)
   'src/react/components/__tests__/client/Query.test.tsx',
 ];

src/__tests__/__snapshots__/exports.ts.snap

@@ -132,6 +132,8 @@ Array [
 exports[`exports of public entry points @apollo/client/errors 1`] = `
 Array [
   "ApolloError",
+  "PROTOCOL_ERRORS_SYMBOL",
+  "graphQLResultHasProtocolErrors",
   "isApolloError",
 ]
 `;

src/__tests__/__snapshots__/graphqlSubscriptions.ts.snap

@@ -2,4 +2,4 @@
 
 exports[`GraphQL Subscriptions should throw an error if the result has errors on it 1`] = `[ApolloError: This is an error]`;
 
-exports[`GraphQL Subscriptions should throw an error if the result has errors on it 2`] = `[ApolloError: This is an error]`;
+exports[`GraphQL Subscriptions should throw an error if the result has protocolErrors on it 1`] = `[ApolloError: cannot read message from websocket]`;

src/__tests__/graphqlSubscriptions.ts

@@ -2,6 +2,7 @@ import gql from 'graphql-tag';
 
 import { ApolloClient } from '../core';
 import { InMemoryCache } from '../cache';
+import { PROTOCOL_ERRORS_SYMBOL } from '../errors';
 import { QueryManager } from '../core/QueryManager';
 import { itAsync, mockObservableLink } from '../testing';
 
@@ -187,22 +188,17 @@ describe('GraphQL Subscriptions', () => {
 
     const obs = queryManager.startGraphQLSubscription(options);
 
-    const promises = [];
-    for (let i = 0; i < 2; i += 1) {
-      promises.push(
-        new Promise<void>((resolve, reject) => {
-          obs.subscribe({
-            next(result) {
-              reject('Should have hit the error block');
-            },
-            error(error) {
-              expect(error).toMatchSnapshot();
-              resolve();
-            },
-          });
-        }),
-      );
-    }
+    const promise = new Promise<void>((resolve, reject) => {
+      obs.subscribe({
+        next(result) {
+          reject('Should have hit the error block');
+        },
+        error(error) {
+          expect(error).toMatchSnapshot();
+          resolve();
+        },
+      });
+    });
 
     const errorResult = {
       result: {
@@ -223,7 +219,7 @@ describe('GraphQL Subscriptions', () => {
     };
 
     link.simulateResult(errorResult);
-    return Promise.all(promises);
+    return Promise.resolve(promise);
   });
 
   it('should call complete handler when the subscription completes', () => {
@@ -261,4 +257,47 @@ describe('GraphQL Subscriptions', () => {
 
     link.simulateResult(results[0]);
   });
+
+  it('should throw an error if the result has protocolErrors on it', () => {
+    const link = mockObservableLink();
+    const queryManager = new QueryManager({
+      link,
+      cache: new InMemoryCache({ addTypename: false }),
+    });
+
+    const obs = queryManager.startGraphQLSubscription(options);
+
+    const promise = new Promise<void>((resolve, reject) => {
+      obs.subscribe({
+        next(result) {
+          reject("Should have hit the error block");
+        },
+        error(error) {
+          expect(error).toMatchSnapshot();
+          resolve();
+        },
+      });
+    });
+
+    const errorResult = {
+      result: {
+        data: null,
+        extensions: {
+          [PROTOCOL_ERRORS_SYMBOL]: [
+            {
+              message: 'cannot read message from websocket',
+              extensions: [
+                {
+                  code: "WEBSOCKET_MESSAGE_ERROR"
+                }
+              ],
+            } as any,
+          ],
+        }
+      },
+    };
+
+    link.simulateResult(errorResult);
+    return Promise.resolve(promise);
+  });
 });

src/core/QueryManager.ts

@@ -32,7 +32,7 @@ import {
   isNonNullObject,
 } from '../utilities';
 import { mergeIncrementalData } from '../utilities/common/incrementalResult';
-import { ApolloError, isApolloError } from '../errors';
+import { ApolloError, isApolloError, graphQLResultHasProtocolErrors } from '../errors';
 import {
   QueryOptions,
   WatchQueryOptions,
@@ -61,6 +61,7 @@ import {
   shouldWriteResult,
   CacheWriteBehavior,
 } from './QueryInfo';
+import { PROTOCOL_ERRORS_SYMBOL, ApolloErrorOptions } from '../errors';
 
 const { hasOwnProperty } = Object.prototype;
 
@@ -936,10 +937,17 @@ export class QueryManager<TStore> {
           this.broadcastQueries();
         }
 
-        if (graphQLResultHasError(result)) {
-          throw new ApolloError({
-            graphQLErrors: result.errors,
-          });
+        const hasErrors = graphQLResultHasError(result);
+        const hasProtocolErrors = graphQLResultHasProtocolErrors(result);
+        if (hasErrors || hasProtocolErrors) {
+          const errors: ApolloErrorOptions = {};
+          if (hasErrors) {
+            errors.graphQLErrors = result.errors;
+          }
+          if (hasProtocolErrors) {
+            errors.protocolErrors = result.extensions[PROTOCOL_ERRORS_SYMBOL];
+          }
+          throw new ApolloError(errors);
         }
 
         return result;
@@ -1235,7 +1243,7 @@ export class QueryManager<TStore> {
       setTimeout(() => concast.cancel(reason));
     });
 
-    let concast: Concast<ApolloQueryResult<TData>>, 
+    let concast: Concast<ApolloQueryResult<TData>>,
         containsDataFromLink: boolean;
     // If the query has @export(as: ...) directives, then we need to
     // process those directives asynchronously. When there are no

src/errors/__tests__/ApolloError.ts

@@ -7,14 +7,26 @@ describe('ApolloError', () => {
       new GraphQLError('Something went wrong with GraphQL'),
       new GraphQLError('Something else went wrong with GraphQL'),
     ];
+    const protocolErrors = [
+      {
+        message: "cannot read message from websocket",
+        extensions: [
+          {
+            code: "WEBSOCKET_MESSAGE_ERROR",
+          },
+        ],
+      }
+    ];
     const networkError = new Error('Network error');
     const errorMessage = 'this is an error message';
     const apolloError = new ApolloError({
       graphQLErrors: graphQLErrors,
+      protocolErrors: protocolErrors,
       networkError: networkError,
       errorMessage: errorMessage,
     });
     expect(apolloError.graphQLErrors).toEqual(graphQLErrors);
+    expect(apolloError.protocolErrors).toEqual(protocolErrors);
     expect(apolloError.networkError).toEqual(networkError);
     expect(apolloError.message).toBe(errorMessage);
   });
@@ -61,6 +73,28 @@ describe('ApolloError', () => {
     expect(messages[1]).toMatch('network error message');
   });
 
+  it('should add both protocol and graphql errors to the message', () => {
+    const graphQLErrors = [new GraphQLError('graphql error message')];
+    const protocolErrors = [
+      {
+        message: "cannot read message from websocket",
+        extensions: [
+          {
+            code: "WEBSOCKET_MESSAGE_ERROR",
+          },
+        ],
+      }
+    ];
+    const apolloError = new ApolloError({
+      graphQLErrors,
+      protocolErrors,
+    });
+    const messages = apolloError.message.split('\n');
+    expect(messages.length).toBe(2);
+    expect(messages[0]).toMatch('graphql error message');
+    expect(messages[1]).toMatch('cannot read message from websocket');
+  });
+
   it('should contain a stack trace', () => {
     const graphQLErrors = [new GraphQLError('graphql error message')];
     const networkError = new Error('network error message');

src/errors/index.ts

@@ -1,10 +1,46 @@
 import '../utilities/globals';
 
-import { GraphQLError } from 'graphql';
+import { GraphQLError, GraphQLErrorExtensions } from 'graphql';
 
 import { isNonEmptyArray } from '../utilities';
 import { ServerParseError } from '../link/http';
 import { ServerError } from '../link/utils';
+import { FetchResult } from "../link/core";
+
+// This Symbol allows us to pass transport-specific errors from the link chain
+// into QueryManager/client internals without risking a naming collision within
+// extensions (which implementers can use as they see fit).
+export const PROTOCOL_ERRORS_SYMBOL: unique symbol = Symbol();
+
+type FetchResultWithSymbolExtensions<T> = FetchResult<T> & {
+  extensions: Record<string | symbol, any>
+};
+
+export interface ApolloErrorOptions {
+  graphQLErrors?: ReadonlyArray<GraphQLError>;
+  protocolErrors?: ReadonlyArray<{
+    message: string;
+    extensions?: GraphQLErrorExtensions[];
+  }>;
+  clientErrors?: ReadonlyArray<Error>;
+  networkError?: Error | ServerParseError | ServerError | null;
+  errorMessage?: string;
+  extraInfo?: any;
+}
+
+export function graphQLResultHasProtocolErrors<T>(
+  result: FetchResult<T>
+): result is FetchResultWithSymbolExtensions<T> {
+  if (result.extensions) {
+    return Array.isArray(
+      (result as FetchResultWithSymbolExtensions<T>).extensions[
+        PROTOCOL_ERRORS_SYMBOL
+      ]
+    );
+  }
+  return false;
+}
+
 
 export function isApolloError(err: Error): err is ApolloError {
   return err.hasOwnProperty('graphQLErrors');
@@ -17,9 +53,14 @@ export function isApolloError(err: Error): err is ApolloError {
 const generateErrorMessage = (err: ApolloError) => {
   let message = '';
   // If we have GraphQL errors present, add that to the error message.
-  if (isNonEmptyArray(err.graphQLErrors) || isNonEmptyArray(err.clientErrors)) {
+  if (
+    isNonEmptyArray(err.graphQLErrors) ||
+    isNonEmptyArray(err.clientErrors) ||
+    isNonEmptyArray(err.protocolErrors)
+  ) {
     const errors = ((err.graphQLErrors || []) as readonly Error[])
-      .concat(err.clientErrors || []);
+      .concat(err.clientErrors || [])
+      .concat((err.protocolErrors || []) as readonly Error[]);
     errors.forEach((error: Error) => {
       const errorMessage = error
         ? error.message
@@ -45,6 +86,10 @@ export class ApolloError extends Error {
   public name: string;
   public message: string;
   public graphQLErrors: GraphQLErrors;
+  public protocolErrors: ReadonlyArray<{
+    message: string;
+    extensions?: GraphQLErrorExtensions[];
+  }>;
   public clientErrors: ReadonlyArray<Error>;
   public networkError: Error | ServerParseError | ServerError | null;
 
@@ -58,20 +103,16 @@ export class ApolloError extends Error {
   // value or the constructed error will be meaningless.
   constructor({
     graphQLErrors,
+    protocolErrors,
     clientErrors,
     networkError,
     errorMessage,
     extraInfo,
-  }: {
-    graphQLErrors?: ReadonlyArray<GraphQLError>;
-    clientErrors?: ReadonlyArray<Error>;
-    networkError?: Error | ServerParseError | ServerError | null;
-    errorMessage?: string;
-    extraInfo?: any;
-  }) {
+  }: ApolloErrorOptions) {
     super(errorMessage);
     this.name = 'ApolloError';
     this.graphQLErrors = graphQLErrors || [];
+    this.protocolErrors = protocolErrors || [];
     this.clientErrors = clientErrors || [];
     this.networkError = networkError || null;
     this.message = errorMessage || generateErrorMessage(this);

src/link/core/types.ts

@@ -21,10 +21,7 @@ export interface ExecutionPatchInitialResult<
   extensions?: TExtensions;
 }
 
-export interface IncrementalPayload<
-  TData,
-  TExtensions,
-> {
+export interface IncrementalPayload<TData, TExtensions> {
   // data and path must both be present
   // https://github.com/graphql/graphql-spec/pull/742/files#diff-98d0cd153b72b63c417ad4238e8cc0d3385691ccbde7f7674bc0d2a718b896ecR288-R293
   data: TData | null;
@@ -48,6 +45,16 @@ export interface ExecutionPatchIncrementalResult<
   extensions?: never;
 }
 
+export interface ApolloPayloadResult<
+  TData = Record<string, any>,
+  TExtensions = Record<string, any>
+> {
+  payload: SingleExecutionResult | ExecutionPatchResult | null;
+  // Transport layer errors (as distinct from GraphQL or NetworkErrors),
+  // these are fatal errors that will include done: true.
+  errors?: ReadonlyArray<Error | string>;
+}
+
 export type ExecutionPatchResult<
   TData = Record<string, any>,
   TExtensions = Record<string, any>